amazon-web-services - Spring Cloud SQS - 轮询间隔

Question

监听 AWS SQS 队列，使用 spring cloud 如下:

@SqsListener(value = "${queue.name}", deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
public void queueListener(String message, @Headers Map<String, Object> sqsHeaders) {
    // code
}

Spring 配置:

<aws-messaging:annotation-driven-queue-listener
    max-number-of-messages="10" wait-time-out="20" visibility-timeout="3600"
    amazon-sqs="awsSqsClient" />

AwsSqsClient:

@Bean
public com.amazonaws.services.sqs.AmazonSQSAsyncClient awsSqsClient() {
    ExecutorService executorService = Executors.newFixedThreadPool(10);
    return new AmazonSQSAsyncClient(new DefaultAWSCredentialsProviderChain(), executorService);
}

这工作正常。

如上代码所示，在 SQS 客户端配置了 10 个线程来处理这些消息。这也工作正常，在任何时间点最多处理 10 条消息。

问题是，我想不出一种方法来控制轮询间隔。默认情况下，一旦所有线程都空闲，spring 会进行轮询。

即考虑以下示例

大约 3 条消息传递到队列

Spring 轮询队列并获得 3 条消息

3 条消息正在处理每条消息大约需要 20 分钟

与此同时，大约有 25 条消息发送到队列。 Spring 不会轮询队列，直到之前传递的所有 3 条消息完成。基本上按照上面的示例，Spring 民意调查仅在 20 分钟后进行，尽管仍有 7 个线程空闲!!

知道我们如何控制这个投票吗？即如果有任何线程空闲，则轮询应该开始，并且不应等到所有线程都空闲时

Answer 1

您的监听器可以将消息加载到您的 Spring 应用程序中，并将它们与 Acknowledgement 一起提交到另一个线程池。和 Visibility对象(如果你想控制两者)。

一旦消息被提交到这个线程池，你的监听器就可以加载更多的数据。您可以通过调整线程池设置来控制并发。

您的监听器的方法签名将类似于以下之一:

@SqsListener(value = "${queueName}", deletionPolicy = SqsMessageDeletionPolicy.NEVER)
public void listen(YourCustomPOJO pojo,
                   @Headers Map<String, Object> headers,
                   Acknowledgment acknowledgment,
                   Visibility visibility) throws Exception {
...... Send pojo to worker thread and return

然后工作线程将确认成功处理

acknowledgment.acknowledge().get();

确保您的 message visibility设置为大于最高处理时间的值(使用一些超时来限制执行时间)。