Spring Cloud Stream Kinesis绑定器关闭时处理中消息失败问题求助

1. 调整Bean销毁顺序，延迟Redis连接工厂销毁 #

通过实现SmartLifecycle接口，让LettuceConnectionFactory的销毁阶段晚于Kinesis消息消费者组件，确保消息处理完成后再销毁Redis连接：

@Configuration
public class RedisConfig {

    @Bean
    public LettuceConnectionFactory lettuceConnectionFactory() {
        LettuceConnectionFactory factory = new LettuceConnectionFactory();
        // 按需配置Redis连接参数
        return factory;
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        // 配置序列化规则等
        return template;
    }

    // 自定义生命周期处理器，控制Redis连接工厂的销毁时机
    @Bean
    public SmartLifecycle redisConnectionLifecycleManager(LettuceConnectionFactory connectionFactory) {
        return new SmartLifecycle() {
            private boolean running = false;

            @Override
            public void start() {
                running = true;
            }

            @Override
            public void stop() {
                // 可在此添加等待消息处理完成的额外逻辑
                connectionFactory.destroy();
                running = false;
            }

            @Override
            public boolean isRunning() {
                return running;
            }

            // 设置更高的phase值，确保晚于Kinesis消费者停止（默认消费者phase通常为0）
            @Override
            public int getPhase() {
                return 100;
            }
        };
    }
}

2. 配置Spring Cloud Stream关闭超时 #

在application.yml中添加Kinesis绑定器的关闭超时配置，给足够时间让正在处理的消息完成：

spring:
  cloud:
    stream:
      kinesis:
        binder:
          shutdown-timeout: 30000 # 30秒，根据实际消息处理耗时调整

3. 订阅errorChannel处理关闭时的异常 #

添加errorChannel的消息处理器，避免Dispatcher has no subscribers异常：

@Configuration
public class ErrorHandlingConfig {

    private static final Logger log = LoggerFactory.getLogger(ErrorHandlingConfig.class);

    @Bean
    public IntegrationFlow errorFlow() {
        return IntegrationFlows.from("errorChannel")
                .handle(message -> {
                    Throwable error = ((ErrorMessage) message).getPayload();
                    log.error("处理关闭阶段的消息错误: ", error);
                })
                .get();
    }
}

4. 监听容器关闭事件，确保消息处理完成后再销毁Redis连接 #

通过监听应用关闭事件，先等待所有Kinesis消费者容器停止，再销毁Redis连接工厂：

@Component
public class ContainerShutdownListener implements ApplicationListener<ContextClosedEvent> {

    private final KinesisMessageChannelBinder binder;
    private final LettuceConnectionFactory connectionFactory;
    private static final Logger log = LoggerFactory.getLogger(ContainerShutdownListener.class);

    public ContainerShutdownListener(KinesisMessageChannelBinder binder, LettuceConnectionFactory connectionFactory) {
        this.binder = binder;
        this.connectionFactory = connectionFactory;
    }

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        try {
            // 遍历所有消费者容器，等待其停止
            binder.getConsumerContainers().forEach(container -> {
                try {
                    container.stop();
                    container.awaitStop(30000); // 最多等待30秒
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("等待容器关闭被中断", e);
                }
            });
        } finally {
            // 确保最后销毁Redis连接工厂
            connectionFactory.destroy();
        }
    }
}

内容的提问来源于stack exchange，提问作者Chintan Radia