弹簧集成处理器

Question

我想定义一个写入 MongoDB 的流，并且只有在成功时才会将 ID 写入 Kafka。我正在使用 JavaDSL，我希望有一个 FlowBuilder 类来定义我的高级管道。我正在寻找能够让我编写流程的功能，例如：

public IntegrationFlow buildFlow() {
   return IntegrationFlows.from(reactiveKafkaConsumerTemplate)
      .process(writeToMongo) // <-- Searching for this kind of function
      .handle(writeToKafka)
      .get();
}

我已经看到Apache Camel works exactly like this，我想知道 Spring Integration 是否也有一个简单而好的解决这个基本问题的方法。

Answer 1

您正在寻找的是publishSubscribeChannel()，它可以拥有多个订阅者。默认情况下，如果没有在频道上配置执行器，下一个订阅者只会在前一个订阅者之后并且只有当这个订阅者成功时才会被调用。

它可能看起来和你用 process() 表达的相似：

public IntegrationFlow buildFlow() {
   return IntegrationFlows.from(reactiveKafkaConsumerTemplate)
      .publishSubscribeChannel(c -> c
                        .subscribe(sf -> sf
                                .handle(MongoDb.reactiveOutboundChannelAdapter())) 
      .handle(writeToKafka)
      .get();
}

另一个选项是gateway()，但是你需要从那里返回一些东西才能继续。在 Spring Integration 中，如果没有回复，则流程将停止。如果没有out，它没有将in 用于out 的概念。