升级到 Spring Integration 5 后不再使用消息

Question

我正在尝试将使用 Spring Integration 4.3 和 Spring Boot 1.6 的项目升级到 Spring Integration 5.1 和 Spring Boot 2.1。以前我有以下配置：

IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "queueName")
                    .id("myId")
                    .autoStartup(autoStartup)
                    .prefetchCount(10)
                    .concurrentConsumers(2)
                    .maxConcurrentConsumers(3)
                    .messageConverter(messageConverter()))
                    .aggregate(a -> a.correlationExpression("payload.entityId")
                                    .releaseExpression("size() eq iterator().next().payload.batchSize")
                                    .sendPartialResultOnExpiry(true)
                                    .groupTimeout(2000)
                                    .expireGroupsUponCompletion(true)
                                    .outputProcessor(myMessageGroupProcessor))
                    .handle(serviceActivatorBean, "myMethod", e -> e.advice(requestHandlerRetryAdviceForIntegrationFlow()))
                    .get();

在升级过程中，我尝试按照文档 here 进行操作，因此将配置更改为：

@Configuration
@EnableAutoConfiguration
@EnableIntegration
public class SpringConfig {

    @Bean(name = "myFlowId")
    public IntegrationFlow myFlow(ConnectionFactory connectionFactory, ServiceActivatorBean serviceActivatorBean,
                                  @Value("${spring.integration.flow.auto-startup:true}") boolean autoStartup,
                                  MyMessageGroupProcessor myMessageGroupProcessor) {
        IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "queueName")
                        .id("myId")
                        .autoStartup(autoStartup)
                        .configureContainer(c -> c.acknowledgeMode(MANUAL)
                            .prefetchCount(10)
                            .concurrentConsumers(2)
                            .maxConcurrentConsumers(3)
                        )
                        .messageConverter(messageConverter()))
                        .aggregate(a -> a.correlationExpression("payload.entityId")
                                        .releaseExpression("size() eq one().payload.batchSize")
                                        .sendPartialResultOnExpiry(true)
                                        .groupTimeout(2000)
                                        .expireGroupsUponCompletion(true)
                                        .outputProcessor(myMessageGroupProcessor))
                        .handle(serviceActivatorBean, "myMethod", e -> e.advice(requestHandlerRetryAdviceForIntegrationFlow()))
                        .get();
    }
}

但是当我发布消息时，集成流程似乎并不received/handled。我没有收到任何错误日志（或任何与此相关的日志，即使我启用了调试日志记录）而且我不太确定从哪里开始调试。我确信消息实际上已发布到 RabbitMQ，所以这不是问题所在。我可能缺少什么？

Answer 1

我的问题实际上不是由于 Spring 集成，而是与 Spring AMQP 的更改有关。以前 "declarables" 可以这样创建：

@Bean
List<Binding> myBinding() {
    return List.of(<binding1>, <binding2>, ..)
}

但在 Spring AMQP 2.1 中应更改为：

@Bean
Declarables myBinding() {
    return new Declarables(List.of(<binding1>, <binding2>, ..))
}

参见文档 here。

对了，我的releaseExpression也错了，应该是size() eq one.payload.batchSize。

升级到 Spring Integration 5 后不再使用消息

Consuming messages no longer works after upgrading to Spring Integration 5

java

spring-integration

spring-boot

spring-integration-dsl