Spring 集成 + SQS - 异常重试不起作用

Question

我正在努力将 Spring 与 AWS SQS 队列集成。

当我用 @ServiceActivator 注释的方法抛出异常时，我遇到了一个问题。在这种情况下，似乎无论如何都会从队列中删除消息。我已经在 SqsMessageDrivenChannelAdapter.

中将 MessageDeletionPolicy 配置为 ON_SUCCESS

这是我的 channel/adapter 配置 https://github.com/sdusza1/spring-integration-sqs/blob/master/src/main/java/com/example/demo/ChannelConfig.java

我已尝试使用 @SqsListener 注释进行同样的操作，但未按预期删除消息。

我在这里创建了一个迷你 Spring 启动应用程序来演示这个问题： https://github.com/sdusza1/spring-integration-sqs

请帮忙:)

Answer 1

你的配置是这样的：

@Bean
public MessageProducerSupport sqsMessageDrivenChannelAdapter() {
    SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(amazonSqs, SQS_QUEUE_NAME);
    adapter.setOutputChannel(inboundChannel());
    adapter.setMessageDeletionPolicy(SqsMessageDeletionPolicy.ON_SUCCESS);
    adapter.setVisibilityTimeout(RETRY_NOTIFICATION_AFTER);
    return adapter;

}

其中inboundChannel是这样的：

 @Bean
    public QueueChannel inboundChannel() {
        return new QueueChannel();
 }

所以，这是一个 队列 ，因此异步和来自该队列的消息由 TaskScheduler 在单独的线程上处理，该线程根据您的 PollerMetadata 配置。在这种情况下，消费者中的任何错误也会被抛入该线程，并且不会达到预期错误处理的 SqsMessageDrivenChannelAdapter。

这在技术上与您的 @SqsListener 体验完全不同，后者真正直接在容器线程上调用，因此应用了错误处理。

或者您需要修改您的逻辑，您希望如何处理该单独线程中的错误，或者只是不在 SqsMessageDrivenChannelAdapter 之后使用 QueueChannel 并让它抛出并处理错误@SqsListener.

情况下的底层 SQS 侦听器容器