如何将助焊剂链到另一个助焊剂/单声道并施加另一个背压？

Question

我有下面的反应代码使用通量在反应堆核心：

Flux.create(sink -> ... /* listens to and receives from external source */ , FluxSink.OverflowStrategy.LATEST)
    .flatMap(map -> redisHashReactiveCommands.hmset(key, map))
    //.flatMap(... //want to store same data async into kafka with its own back pressure handling)
    .subscribeOn(Schedulers.parallel())
    .doOnNext(s -> log.debug("Redis consumed. Result -> {}", s))
    .doOnComplete(() -> log.debug("On completed."))
    .doOnError(exception -> log.error("Error occurred while consuming message", exception))
    .subscribe();

正如您所看到的，对于我的流程的外部源（FluxSink.OverflowStrategy.Latest），我对此进行了背压处理。但是，我还想为我的进程配置redis（redishashreactiveCommands.hmset（key,map））的背压，因为它对我的进程来说可能是一个比外部源更大的瓶颈。我希望我需要为redis部分创建另一个flux并将它与这个flux链接起来，但是我如何实现这一点，因为.FlatMap只工作于单个项而不是项流？

同样，我也想存储相同的发射项目到Kafka，但链接FlapMap的似乎不起作用…有没有一种简单的方法将所有这些链接在一组函数调用中（外部源->我的流程，我的流程->redis，我的流程->kafka)？

Answer 1

如果您对主序列中的结果对象不感兴趣，可以在FlatMap中合并两个保存。您还必须在flatMap中移动订阅和日志，以便将它们放在内部保存发布服务器上：

Flux.create(sink -> ... /* listens to and receives from external source */ , FluxSink.OverflowStrategy.LATEST)
    .flatMap(map -> Mono.when(
        redisHashReactiveCommands.hmset(key, map)
            .subscribeOn(Schedulers.parallel())
            .doOnNext(s -> log.debug("Redis consumed. Result -> {}", s)),

        kafkaReactiveCommand.something(map)
            .subscribeOn(Schedulers.parallel())
            .doOnNext(s -> log.debug("Kafka consumed. Result -> {}", s)),
    ))
    //... this results in a Mono<Void>
    .doOnComplete(() -> log.debug("Both redis and kafka completed."))
    .doOnError(exception -> log.error("Error occurred while consuming message", exception))
    .subscribe();

或者，如果您确信两个进程都发出一个结果元素或一个错误，您可以通过将when替换为zip来将这两个结果合并到一个tuple2中。