了解 SqsMessageDrivenChannelAdapter 行为答案

【问题标题】：Understanding SqsMessageDrivenChannelAdapter behaviour了解 SqsMessageDrivenChannelAdapter 行为
【发布时间】：2019-10-08 11:09:37
【问题描述】：

我试图了解SqsMessageDrivenChannelAdapter 的行为以解决内存问题。上游系统在aws-sqs-queue中转储了数千条消息，所有消息都立即被SqsMessageDrivenChannelAdapter接收。在 AWS 控制台上，我看不到队列中的任何可用消息。然后SqsMessageProcesser 每 5 秒处理 1 条消息。

这是日志：

2019-05-21 17:28:18 INFO SQSMessageProcessor:88 - ---里面 sqsMessageProcesser--- 2019-05-21 17:28:23 信息 SQSMessageProcessor:88 ---- sqsMessageProcesser 内部--- 2019-05-21 17:28:28 INFO SQSMessageProcessor:88 ----里面 sqsMessageProcesser--- 2019-05-21 17:28:33 信息 SQSMessageProcessor:88 ---- sqsMessageProcesser 内部--- 2019-05-21 17:28:38 INFO SQSMessageProcessor:88 ----里面 sqsMessageProcesser--- ......................

这是否意味着当SqsMessageProcesser 每 5 秒处理 1 条消息时，in-channel 的（服务器）内存中保存着数千条消息？

每个数据库事务大约需要 5 秒，目前我们在 PRD 上面临“内存不足”问题。

如果我在QueueChannel 和setMaxNumberOfMessages 上为SqsMessageDrivenChannelAdapter 设置容量会有帮助吗？如果是，是否有计算这些值的标准方法？


  @Bean(name = "in-channel")
  public PollableChannel sqsInputChannel() {
    return new QueueChannel();
  }

  @Autowired
  private AmazonSQSAsync amazonSqs;

  @Bean
  public MessageProducer sqsMessageDrivenChannelAdapterForItems() {

    SqsMessageDrivenChannelAdapter adapter =
        new SqsMessageDrivenChannelAdapter(amazonSqs, "aws-sqs-queue");
    adapter.setOutputChannelName("in-channel");
    return adapter;
  }

  @ServiceActivator(inputChannel = "in-channel",
      poller = @Poller(fixedRate = "5000" , maxMessagesPerPoll = "1"))
  public void sqsMessageProcesser(Message<?> receive) throws ProcesserException {
  logger.info("--- inside sqsMessageProcesser---")
  // db transactions.
}

【问题讨论】：

标签： spring-integration spring-cloud-aws spring-integration-aws

【解决方案1】：

实际上，为消息驱动的通道适配器放置QueueChannel 是一种反模式。后者已经是异步的并且基于一些任务调度。因此，将消费消息从源转移到内存队列肯定会带来一些麻烦。

您应该考虑改为使用直接通道，并让 SQS 消费线程被阻塞，直到您的 sqsMessageProcesser 完成其工作。这样您就可以保证不会丢失任何数据。

【讨论】：