带有SQS和Reactive的Spring集成流DSL

Question

如何使用DSL为以下步骤设置无功流：

1. 使用SqsMessageDrivenChannelAdapter接收SQS消息
2. 使用验证方法验证Json消息[JsonSchemaValidator类]
3. 将json转换为对象
4. 将对象传递给服务激活器（BusinessService:business logic，state machine）
5. 持久化对象R2DBC出站适配器

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

在上述示例中，创建了专用流，返回发布者，在测试中，发布者被订阅。然而，当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：使用executor通道将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集成中针对反应流所做的工作：https://docs.spring.io/spring-integration/docs/current/reference/html/reactive-streams.html#reactive-溪流

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

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

您可以考虑在流的endpoint之间放置一个FLuxMessageChannel，但它对您的用例仍然没有意义。它不会像您期望的那样完全反应性，仅仅因为org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer没有阻塞消费者线程以准备来自下游的反压力。

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

正如我所说：您可以尝试放置一个通道（通道-