在 Spring Integration 中用于持久性传递到 AMQP 代理的消息存储

Question

我正在尝试构建集成流，这将防止消息在传递到 AMQP 代理 (rabbitMQ) 期间丢失。 在代理停止的情况下，我看到了一些令我意想不到的行为：

1. 失败的消息正在保存到消息存储中，但不会保存很长时间。此流程不会等待代理可用，它会从消息存储中提取消息，即使代理仍处于停止状态
2. 如果成功重新启动 rabbitmq，消息存储中的记录（如果它们仍然存在）不会被传递到队列。

请帮助我进行调查。代码示例：

 @Bean
public MessageChannel messageStoreBackedChannel() {
    return new QueueChannel(
            new MessageGroupQueue(jdbcChannelMessageStore(), "Group_ID")
    );
}

 @Bean
public IntegrationFlow someFlow() {
    return IntegrationFlows
            .from("messageStoreBackedChannel")
            .channel("amqpMessageChannel")
            .get();
}

@Bean
public IntegrationFlow jmsExtractFlow(EntityManagerFactory entityManagerFactory) {
    return IntegrationFlows
            .from("amqpMessageChannel")
            .handle(message -> System.out.println(message.getPayload()))
            .get();
}


@Bean
public MessageChannel amqpMessageChannel() {
    return new PollableAmqpChannel("amqpMessageChannel", amqpTemplate);
}

@Bean
public JdbcChannelMessageStore jdbcChannelMessageStore() {
    var jdbcChannelMessageStore = new JdbcChannelMessageStore(dataSource);
    jdbcChannelMessageStore.setChannelMessageStoreQueryProvider(new PostgresChannelMessageStoreQueryProvider());

    return jdbcChannelMessageStore;
}

@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {
    PollerMetadata pollerMetadata = new PollerMetadata();
    pollerMetadata.setTrigger(new PeriodicTrigger(10));
    return pollerMetadata;
}

Answer 1

考虑将.from("messageStoreBackedChannel").channel("amqpMessageChannel") 之间的端点配置为transactional()。

类似这样的：

.from("messageStoreBackedChannel")
.bridge(e -> e.poller(p -> p.fixedDelay(10).transactional()))
.channel("amqpMessageChannel")

因此，每当传递到amqpMessageChannel 失败时，事务将回滚，失败的消息将返回到存储区，直到下一次轮询。

当然，当您在连接到 RabbitMQ 时遇到错误时，您可以停止 bridge 端点。但是你怎么能确定那个连接又回来了呢？..