数量并行处理简单队列服务（SQS）答案

【问题标题】：Amount parallel processing Simple Queue Service (SQS)数量并行处理简单队列服务（SQS）
【发布时间】：2018-03-23 15:31:12
【问题描述】：

我正在使用 Spring Cloud 来使用简单队列服务 (SQS)。我有以下并行处理配置：

@Bean
public SimpleAsyncTaskExecutor simpleAsyncTaskExecutor() {
    SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
    simpleAsyncTaskExecutor.setConcurrencyLimit(50);
    return simpleAsyncTaskExecutor;
}

@Bean
public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(
        SimpleAsyncTaskExecutor simpleAsyncTaskExecutor) {

    SimpleMessageListenerContainerFactory factory = new SimpleMessageListenerContainerFactory();
    factory.setAutoStartup(true);
    factory.setTaskExecutor(simpleAsyncTaskExecutor);
    factory.setWaitTimeOut(20);
    factory.setMaxNumberOfMessages(10);

    return factory;
}

我需要在 50 个线程中处理 50 条消息（在 bean SimpleAsyncTaskExecutor 中配置），但只并行处理 10 条消息（从 SQS 返回的 maxNumberOfMessages）

如何处理 50 条消息而不是 10 条消息？

【问题讨论】：

那么它是在 50 个线程中（并行）处理 10 条消息，还是总共处理 10 条消息？您已经设置了SimpleMessageListenerContainerFactory.setMaxNumberOfMessages(10);，所以如果它同时处理 10 条消息乘以 50 个线程，那么将这个数字增加到 50？ edit 没关系，我在文档中看到，最大消息公关。请求为 10。如果遇到瓶颈，则需要合并更多线程。
在文档中，每个请求返回的最大消息数为 10。因此它在 50 个线程中处理 10 条消息。当 10 条消息的处理结束时，会发出一个新的请求。我想在 SQS 队列中发出 5 个请求，这将导致在 50 个线程中处理 50 条消息。

标签： java spring spring-cloud amazon-sqs

【解决方案1】：

我找到了解决办法。

需要将方法注解@Async，将deletionPolicy改为NEVER，并在结束执行时删除消息。

这样，队列消费将遵循配置的线程数。例如，如果您有 50 个线程，将在 SQS 队列中发出 5 个请求（每个请求 10 条消息），因此总共并行处理 50 条消息。

代码如下所示：

@Async
@SqsListener(value = "sqsName", deletionPolicy = SqsMessageDeletionPolicy.NEVER)
public void consume(String message, Acknowledgment acknowledgment) throws InterruptedException, ExecutionException {

    //your code

    acknowledgment.acknowledge().get(); //To delete message from queue
}

【讨论】：

【解决方案2】：

我不会过多关注特定数字（例如 50 个线程的 50 条消息）。尝试对其进行性能测试（构建一些东西以将高峰时段的预期消息数量推送到队列，并让您的服务处理它们，看看它是否会出现瓶颈）。

根据您的实际问题，您不能。 AWS SQS 根本不支持获取超过 10 条消息。要求。请参阅http://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html 以供参考。（在第一段）。

【讨论】：

我的邮件量很大。所以一次10条消息是不够的。在文档中，每个请求实际上只返回 10 条消息，但我想向 SQS 队列发出多个请求。例如，我想向 SQS 发出 5 次请求，总共产生 50 条消息，其中这 50 条将在 50 个 threds 中处理。
啊，我明白你的意思了。我认为执行程序正在同时执行轮询请求，但我看到它用于处理实际消息的容器中。我没有关于 spring-cloud-aws 的知识来即时提出适当的建议，而我自己也没有摆弄它。对不起。