Spring 使用 SQS 和 Reactive 的集成流 DSL

Question

如何使用 DSL 为以下步骤设置反应流：

1. 使用 SqsMessageDrivenChannelAdapter
接收 SQS 消息
2. 使用 validate 方法验证 Json 消息 [JsonSchemaValidator class]
3. 将 json 转换为对象
4. 将对象传递给服务激活器（BusinessService：业务逻辑、状态机）
5. 保留对象 R2DBC 出站适配器

我在看这个：https://github.com/spring-projects/spring-integration/blob/master/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java

在上面的示例中，创建了 return 和 Publisher 的专用流，并且在测试中发布者是 subscribed。但是，当 SqsMessageDrivenChannelAdapter 将消息引入频道时，我的流程将被触发。

对于上述步骤 1 到 5 的场景，如何实现反应流配置？

更新：添加了示例代码

   @Bean
    public IntegrationFlow importFlow()  {
        IntegrationFlows.from(sqsInboundChannel())
                .handle((payload, messageHeaders) -> jsonSchemaValidator.validate(payload.toString()))
                .transform(Transformers.fromJson(Entity.class))
                .handle((payload, messageHeaders) ->businessService.process((Entity) payload))
                .handle(
                        Jpa.outboundAdapter(this.entityManagerFactory)
                                .entityClass(Entity)
                                .persistMode(PersistMode.PERSIST),
                        ConsumerEndpointSpec::transactional)
                .get();
    }

    @Bean
    public MessageProducer sqsMessageDrivenChannelAdapter() {
        SqsMessageDrivenChannelAdapter sqsMessageDrivenChannelAdapter =
                new SqsMessageDrivenChannelAdapter(asyncSqsClient, queueName);
        sqsMessageDrivenChannelAdapter.setAutoStartup(true);
        sqsMessageDrivenChannelAdapter.setOutputChannel(sqsInboundChannel());
        return sqsMessageDrivenChannelAdapter;
    }

    @Bean
    public MessageChannel sqsInboundChannel() {
        return MessageChannels.flux().get();
    }

更新 2：使用执行程序通道将 JPA 移动到 diff 线程

   @Bean
    public IntegrationFlow importFlow()  {
        IntegrationFlows.from(sqsInboundChannel())
                .handle((payload, messageHeaders) -> jsonSchemaValidator.validate(payload.toString()))
                .transform(Transformers.fromJson(Entity.class))
                .handle((payload, messageHeaders) ->businessService.process((Entity) payload))
                .channel(persistChannel())
                .handle(
                        Jpa.outboundAdapter(this.entityManagerFactory)
                                .entityClass(Entity)
                                .persistMode(PersistMode.PERSIST),
                        ConsumerEndpointSpec::transactional)
                .get();
    }

    @Bean
    public MessageProducer sqsMessageDrivenChannelAdapter() {
        SqsMessageDrivenChannelAdapter sqsMessageDrivenChannelAdapter =
                new SqsMessageDrivenChannelAdapter(asyncSqsClient, queueName);
        sqsMessageDrivenChannelAdapter.setAutoStartup(true);
        sqsMessageDrivenChannelAdapter.setOutputChannel(sqsInboundChannel());
        return sqsMessageDrivenChannelAdapter;
    }

    @Bean
    public MessageChannel sqsInboundChannel() {
        return MessageChannels.flux().get();
    }

    @Bean
    public MessageChannel persistChannel() {
        return MessageChannels.executor(Executors.newCachedThreadPool()).get();
    }

Answer 1

您可能需要让自己更加熟悉我们目前在 Spring 中对 Reactive Streams 的了解：https://docs.spring.io/spring-integration/docs/current/reference/html/reactive-streams.html#reactive-streams

您在该测试中显示的示例 class 与您的用例完全不相关。在该测试中，我们尝试涵盖我们在 Spring 集成中公开的一些 API，有点像单元测试。与整个流程无关

您的用例实际上只是一个完整的黑盒流程，从 SQS 侦听器开始到 R2DBC 结束。因此，在您的流程中没有必要尝试将它的一部分转换为 Publisher 然后将其带回流程的另一部分：您不会跟踪一些方式并订阅该 Publisher 你自己。

您可以考虑在流程的端点之间放置一个 FluxMessageChannel，但它对您的用例仍然没有意义。它不会像您预期的那样完全反应，只是因为 org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer 没有阻塞消费者线程以准备好承受来自下游的背压。

您的流程中唯一真正具有反应性的部分是 R2DBC 出站通道适配器，但它可能不会给您带来太多价值，因为数据源不是反应性的。

正如我所说：您可以尝试在 SqsMessageDrivenChannelAdapter 定义之后放置一个 channel(channels -> channels.flux()) 以从该点开始反应流。同时，您应该尝试将 maxNumberOfMessages 设置为 1 以尝试使其在从 SQS 拉下一条消息之前等待空闲 space。