Spring Cloud SQS 消费阻塞，直到所有消息处理完毕

Question

我们正在使用Spring Cloud AWS 与 SQS 进行交互。我们使用@SqsListener 注释从我们的队列中提取消息。我们有deletionPolicy = NEVER，这意味着我们手动确认我们选择的所有消息。

我们的问题是SimpleMessageListenerContainer（处理来自队列的消息的处理）等待所有工作线程完成，然后再从队列中挑选更多消息。

换句话说，我们看到的是这样的：

• 从队列中拉出 10 条消息。
• 启动 10 个线程来完成工作。
• 其中一个正在工作的线程在缓慢的 IO 调用中被阻塞。
• 现在阻止应用程序从队列中获取更多消息，因此根本无法执行更多工作，直到缓慢的调用完成。

我们可以看到SimpleMessageListenerContainer.AsynchronousMessageListener中的代码负责这个

@Override
public void run() {
    while (isQueueRunning()) {
        try {
            ReceiveMessageResult receiveMessageResult = getAmazonSqs().receiveMessage(this.queueAttributes.getReceiveMessageRequest());
            CountDownLatch messageBatchLatch = new CountDownLatch(receiveMessageResult.getMessages().size());
            for (Message message : receiveMessageResult.getMessages()) {
                if (isQueueRunning()) {
                    MessageExecutor messageExecutor = new MessageExecutor(this.logicalQueueName, message, this.queueAttributes);
                    getTaskExecutor().execute(new SignalExecutingRunnable(messageBatchLatch, messageExecutor));
                } else {
                    messageBatchLatch.countDown();
                }
            }
            try {
                messageBatchLatch.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        } catch (Exception e) {
            getLogger().warn("An Exception occurred while polling queue '{}'. The failing operation will be " +
                    "retried in {} milliseconds", this.logicalQueueName, getBackOffTime(), e);
            try {
                //noinspection BusyWait
                Thread.sleep(getBackOffTime());
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

理想情况下，我们希望消息侦听器不断从队列中挑选消息进行处理。

我们似乎无法实现自己的MessageListenerContainer，因为AbstractMessageListenerContainer 是本地包。

有没有办法解决这个问题？

Answer 1

保持消息轮询线程的是messageBatchLatch.await() 语句。似乎只需卸下闩锁就可以了。比如：

@Override
public void run() {
    while (isQueueRunning()) {
        try {
            ReceiveMessageResult receiveMessageResult = getAmazonSqs().receiveMessage(this.queueAttributes.getReceiveMessageRequest());
            for (Message message : receiveMessageResult.getMessages()) {
                if (isQueueRunning()) {
                    MessageExecutor messageExecutor = new MessageExecutor(this.logicalQueueName, message, this.queueAttributes);
                    getTaskExecutor().execute(new SignalExecutingRunnable(messageExecutor));
                }
            }
        } catch (Exception e) {
            getLogger().warn("An Exception occurred while polling queue '{}'. The failing operation will be " +
                    "retried in {} milliseconds", this.logicalQueueName, getBackOffTime(), e);
            try {
                //noinspection BusyWait
                Thread.sleep(getBackOffTime());
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

如果您的 TaskExecutor 实现，这将起作用： - 有一个固定大小的线程池 - 当execute 函数被调用并且没有线程可用时阻塞。

这是大多数实现的工作方式，但值得检查一下。