Spring Integration Jdbc ChannelMessageStore失败消息重轮询及重试丢弃问题

问题1：Poller无法自动拾取失败消息（仅服务重启后生效） #

核心原因是失败消息被持续锁定，导致Poller无法再次获取。解决方式如下：

1. 配置JdbcChannelMessageStore的锁超时与缓存设置

   @Bean
   public JdbcChannelMessageStore jdbcChannelMessageStore(DataSource dataSource) {
       JdbcChannelMessageStore store = new JdbcChannelMessageStore(dataSource);
       store.setTablePrefix("INT_"); // 匹配你创建的消息表前缀
       store.setLockTimeout(5000); // 设置锁超时时间（单位：ms），超时后自动释放锁
       store.setUsingIdCache(false); // 关闭ID缓存，确保Poller每次都能查询最新的未锁定消息
       return store;
   }
   

2. 确保Poller绑定事务管理器，异常回滚时释放锁

   @Bean
   public PollerMetadata poller(PlatformTransactionManager transactionManager) {
       return Pollers.fixedRate(1000)
               .transactional(transactionManager) // 绑定事务，回滚时自动释放消息锁
               .get();
   }
   

3. 检查消息处理方法的事务配置，确保异常触发回滚
   在消息处理类的方法上添加事务注解，明确回滚条件：

   @Transactional(rollbackFor = Exception.class)
   public void handleMessage(Message<?> message) {
       // 你的消息处理逻辑
   }
   

问题2：设置最大重试次数，耗尽后丢弃消息 #

通过RequestHandlerRetryAdvice配置重试规则与耗尽后的处理策略：

1. 配置重试通知Bean

   @Bean
   public RequestHandlerRetryAdvice retryAdvice() {
       RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
       RetryTemplate retryTemplate = new RetryTemplate();
       
       // 设置最大重试次数为3次
       SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
       retryPolicy.setMaxAttempts(3);
       retryTemplate.setRetryPolicy(retryPolicy);
       
       // 重试耗尽后的恢复策略：直接丢弃（可替换为发送死信队列等逻辑）
       advice.setRecoveryCallback(context -> {
           System.err.println("消息重试耗尽，已丢弃：" + context.getAttribute("message"));
           return null; // 返回null表示丢弃消息
       });
       
       advice.setRetryTemplate(retryTemplate);
       return advice;
   }
   

2. 在消息处理端点绑定重试通知

   @Bean
   public IntegrationFlow persistentQueueFlow(JdbcChannelMessageStore messageStore, 
                                             PollerMetadata poller, 
                                             RequestHandlerRetryAdvice retryAdvice) {
       return IntegrationFlows.from(MessageChannels.queue("persistentQueue", messageStore)
                       .poller(poller))
               .handle("messageHandler", "handleMessage", config -> config.advice(retryAdvice))
               .get();
   }
   

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