Spring Integration SQS监听队列未触发集成流问题排查求助

问题描述 #

使用Spring Integration Flow结合SQS监听队列，预期接收消息后触发集成流，但测试时消息接收后未触发任何集成流。已根据Artem建议更新配置，现提供相关配置代码请求排查。

现有配置 #

SQS适配器配置 #

@Bean
public MessageProducerSupport sqsMessageDrivenChannelAdapter() {
  SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(amazonSQSAsync, "Main");
  adapter.setOutputChannel(inputChannel());
  adapter.setAutoStartup(true);
  adapter.setMessageDeletionPolicy(SqsMessageDeletionPolicy.NEVER);
  adapter.setMaxNumberOfMessages(10);
  adapter.setVisibilityTimeout(5);
  return adapter;
}

队列通道配置 #

@Bean
public MessageChannel inputChannel() {
  return new DirectChannel();
}

主集成流触发配置 #

@Bean
public IntegrationFlow inbound() {
  return IntegrationFlows.from("inputChannel").transform(i -> "TEST_FLOW").get();
}

排查步骤 #

• 验证SQS客户端有效性：确保amazonSQSAsync Bean配置了正确的AWS密钥、区域，且能正常连接SQS服务。可通过调用amazonSQSAsync.listQueues()测试连接是否正常。
• 核对队列名称：确认AWS控制台中存在名为"Main"的队列，注意名称大小写完全匹配。
• 开启日志排查：开启Spring Integration和AWS SDK的DEBUG级日志，查看是否有SQS消息接收日志、通道消息发送日志，以及集成流启动日志，定位消息流转节点。
• 检查通道消息流转：给inputChannel添加拦截器，验证消息是否被发送到通道：

  @Bean
  public MessageChannel inputChannel() {
      DirectChannel channel = new DirectChannel();
      channel.addInterceptor(new ChannelInterceptor() {
          @Override
          public Message<?> preSend(Message<?> message, MessageChannel channel) {
              System.out.println("消息进入inputChannel: " + message.getPayload());
              return message;
          }
      });
      return channel;
  }
  

• 确认适配器启动状态：应用启动后，通过Spring上下文获取sqsMessageDrivenChannelAdapter Bean，调用isRunning()检查是否正常启动。
• 添加集成流执行日志：在transform方法中添加日志，验证集成流是否被触发：

  @Bean
  public IntegrationFlow inbound() {
      return IntegrationFlows.from("inputChannel")
              .transform(i -> {
                  System.out.println("集成流已触发，接收 payload: " + i);
                  return "TEST_FLOW";
              })
              .get();
  }
  

• 临时调整消息删除策略：将SqsMessageDeletionPolicy.NEVER改为SqsMessageDeletionPolicy.ON_SUCCESS，排除手动删除逻辑对流程的干扰。
• 检查可见性超时设置：当前visibilityTimeout(5)为5秒，若消息处理耗时超过该值，消息会重新入队，但不影响集成流触发，可临时调大值测试。

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