如何在 Spring Integration Reactor 中创建反应式入站通道适配器

Question

我想了解如何为 Spring 与 Reactor 核心的集成创建 reactive 通道适配器。我从其他论坛I've read that this Mongo DB reactive adapter can be a good example 了解到，但它包含许多特定于 Mongo 领域的类。

我已阅读文档的 Reactive 部分，我发现需要实现 MessageProducerSupport，但从代码示例看来，需要实现一个扩展 MessageProducerSpec 的类和调用第一个。有人可以举一个最基本的用法示例并解释创建这样一个通道适配器的真正需求吗？我知道我应该做的是：

public IntegrationFlow buildPipe() {
   return IntegrationFlows.from(myMessageProducerSpec)
      .handle(reactiveMongoDbStoringMessageHandler, "handleMessage")
      .handle(writeToKafka)
      .get();
}

Answer 1

MessageProducerSpec 用于 Java DSL。它与通道适配器的低级逻辑无关。如果你有一个MessageProducerSupport，那么这个就足够你在流定义中使用了：

/**
 * Populate the provided {@link MessageProducerSupport} object to the {@link IntegrationFlowBuilder} chain.
 * The {@link org.springframework.integration.dsl.IntegrationFlow} {@code startMessageProducer}.
 * @param messageProducer the {@link MessageProducerSupport} to populate.
 * @return new {@link IntegrationFlowBuilder}.
 */
public static IntegrationFlowBuilder from(MessageProducerSupport messageProducer) {

查看更多关于 Java DSL 中任意通道适配器使用的文档：https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-dsl-protocol-adapters

但又一次：忘记用于通道适配器的 Java DSL API。首先实现该通道适配器。是的，反应式MessageProducerSupport 必须在其doStart() 实现中使用subscribeToPublisher()。从源系统构建Flux 的其余逻辑取决于您和您要依赖的库。

还有ReactiveRedisStreamMessageProducer和ZeroMqMessageProducer，但我不能说他们的代码比上面提到的MongoDbChangeStreamMessageProducer更容易消化。