使用 Spring Integration with Project Reactor 时的主要处理流程编程方法

Question

我想定义一个使用 Reactor Kafka 消耗 kafka 并写入 MongoDB 的流，并且只有在成功时才会将 ID 写入 Kafka。我正在使用带有 Spring Integration JavaDSL 的 Project Reactor，并且我希望有一个 FlowBuilder 类来定义我的高级管道。我目前有以下方向：

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

我见过docs that there is a support for a different approach, that also works with Project Reactor。这种方法不包括使用IntegrationFlows。这看起来像这样：

@MessagingGateway
public static interface TestGateway {

    @Gateway(requestChannel = "promiseChannel")
    Mono<Integer> multiply(Integer value);

    }

        ...

    @ServiceActivator(inputChannel = "promiseChannel")
    public Integer multiply(Integer value) {
            return value * 2;
    }

        ...

    Flux.just("1", "2", "3", "4", "5")
            .map(Integer::parseInt)
            .flatMap(this.testGateway::multiply)
            .collectList()
            .subscribe(integers -> ...);

我想知道在使用这两个库时更推荐的处理方式。我想知道如何在第二个示例中使用 Reactive MongoDB 适配器。如果没有 IntegrationFlows 包装器，我不确定第二种方法是否可行。

Answer 1

@MessagingGateway 是为高级最终用户 API 设计的，以尽可能地将消息隐藏在下面。因此，当您开发目标服务的逻辑时，目标服务没有任何消息传递抽象。

可以使用IntegrationFlow 中的这样一个接口适配器，您应该将其视为常规服务激活器，因此它看起来像这样：

.handle("testGateway", "multiply", e -> e.async(true))

async(true) 使这个服务激活器订阅返回的Mono。您可以省略它，然后您自己在下游订阅它，因为这个 Mono 将成为流中下一条消息的 payload。

如果你想要相反的东西：从Flux 调用IntegrationFlow，就像flatMap()，然后考虑使用流定义中的toReactivePublisher() 运算符返回Publisher<?> 并声明它作为一颗豆子。在这种情况下，最好不要使用 MongoDb.reactiveOutboundChannelAdapter()，而只使用 ReactiveMongoDbStoringMessageHandler 让它返回的 Mono 传播到 Publisher。

另一方面，如果您希望 @MessagingGateway 与 Mono 返回，但仍从中调用 ReactiveMongoDbStoringMessageHandler，则将其声明为 bean 并用 @ServiceActivator 标记它。

我们还有一个ExpressionEvaluatingRequestHandlerAdvice 来捕获特定端点上的错误（或成功）并分别处理它们：https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#expression-advice

我想你要找的是这样的：

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

注意.handle(reactiveMongoDbStoringMessageHandler) - 这与MongoDb.reactiveOutboundChannelAdapter() 无关。因为这个将ReactiveMessageHandler 包装成ReactiveMessageHandlerAdapter 用于自动订阅。您需要的是看起来更像是您希望将 Mono<Void> 返回到您自己的控制中，因此您可以将其用作您的 writeToKafka 服务的输入，并自己在那里订阅并按照您的解释处理成功或错误。关键是，使用 Reactive Stream，我们无法提供命令式错误处理。该方法与任何异步 API 使用相同。因此，对于 Reactive Streams，我们也将错误发送到 errorChannel。

我们可能可以使用returnMono(true/false) 之类的东西改进MongoDb.reactiveOutboundChannelAdapter()，让像您这样的用例开箱即用。