在使用包括重试的流量时顺序调用非阻塞操作

Question

    Flux<ReceiverRecord<Integer, DataDocument>> messages = Flux.defer(() -> {
        KafkaReceiver<Integer, DataDocument> receiver = KafkaReceiver.create(options);
        return receiver.receive();
    });

    messages.map(this::transformToOutputFormat)
            .map(this::performAction)
            .flatMapSequential(receiverRecordMono -> receiverRecordMono)
            .doOnNext(record -> record.receiverOffset().acknowledge())
            .doOnError(error -> logger.error("Error receiving record", error))
            .retryBackoff(100, Duration.ofSeconds(5), Duration.ofMinutes(5))
            .subscribe();

正如你所看到的，我所做的是：从Kafka那里获取消息，将其转换成一个意在新目的地的对象，然后将其发送到目的地，然后确认偏移量以标记消息为已消费和已处理。以与从Kafka消费的消息相同的顺序确认偏移量是非常关键的，这样我们就不会将偏移量移动到未完全处理的消息之外（包括将一些数据发送到目的地）。因此，我使用FlatMapSequential来确保这一点。

为了简单起见，我们假设TransformToOutputFormat()方法是一个标识转换。

public ReceiverRecord<Integer, DataDocument> transformToOutputFormat(ReceiverRecord<Integer, DataDocument> record) {
    return record;
}

performAction()方法需要通过网络执行某些操作，例如调用HTTP REST API。因此适当的API返回一个Mono，这意味着需要订阅链。此外，我需要此方法返回receiverrecord，以便可以在上面的flatMapSequential（）运算符中确认偏移量。因为我需要订阅的单声道，所以我使用上面的FlatMapSequential。如果不是，我可以使用映射。

public Mono<ReceiverRecord<Integer, DataDocument>> performAction(ReceiverRecord<Integer, DataDocument> record) {
    return Mono.just(record)
            .flatMap(receiverRecord ->
                    HttpClient.create()
                            .port(3000)
                            .get()
                            .uri("/makeCall?data=" + receiverRecord.value().getData())
                            .responseContent()
                            .aggregate()
                            .asString()
            )
            .retryBackoff(100, Duration.ofSeconds(5), Duration.ofMinutes(5))
            .then(Mono.just(record));

我在这个方法中有两个相互冲突的需求：1。订阅进行HTTP调用的链2。返回ReceiverRecord

使用flatMap（）意味着返回类型更改为Mono。在同一位置使用doOnNext（）将保留链中的ReceiverRecord，但不允许自动订阅HttpClient响应。

我无法在asstring()之后添加.subscribe()，因为我想等到完全接收到HTTP响应后再确认偏移量。

结果，我需要作弊并从方法范围返回record对象。

另一件事是在PerformAction内重试时，它会切换线程。由于flatMapSequential（）急切地订阅外部通量中的每个Mono，这意味着虽然可以按顺序保证对偏移量的确认，但我们不能保证PerformAction中的HTTP调用将按相同的顺序执行。

所以我有两个问题。

null

Answer 1

这里是我想出的解决办法。

Flux<ReceiverRecord<Integer, DataDocument>> messages = Flux.defer(() -> {
    KafkaReceiver<Integer, DataDocument> receiver = KafkaReceiver.create(options);
    return receiver.receive();
});

messages.map(this::transformToOutputFormat)
        .delayUntil(this::performAction)
        .doOnNext(record -> record.receiverOffset().acknowledge())
        .doOnError(error -> logger.error("Error receiving record", error))
        .retryBackoff(100, Duration.ofSeconds(5), Duration.ofMinutes(5))
        .subscribe();

我没有使用flatMapSequential订阅performAction Mono并保留序列，而是延迟了从Kafka接收器请求更多消息的请求，直到执行该操作。这就实现了我所需要的一次一次的处理。

因此，performAction不需要返回ReceiverRecord的单声道。我还将其简化为：

public Mono<String> performAction(ReceiverRecord<Integer, DataDocument> record) {
    HttpClient.create()
        .port(3000)
        .get()
        .uri("/makeCall?data=" + receiverRecord.value().getData())
        .responseContent()
        .aggregate()
        .asString()
        .retryBackoff(100, Duration.ofSeconds(5), Duration.ofMinutes(5));
}