java - 带有ThreadpoolExecutor的SQSListener

在下面的示例中，我将最大和核心池大小设置为1。但是，未处理任何消息。当我启用调试日志时，我能够看到从SQS中提取的消息，但是我猜它没有被处理/删除。但是，当我将核心和最大池大小增加到2时，消息似乎已被处理。

编辑

我相信Spring可能为接收方分配了一个线程，该线程从队列中读取数据，因此无法为正在处理消息的监听器分配线程。当我将corepoolsize增加到2时，我看到消息正在从队列中读取。当我添加另一个监听器(用于死信队列)时，遇到了相同的问题-2个线程不足，因为未处理消息。当我将corepoolsize增加到3时，它开始处理消息。我假设在这种情况下，分配了1个线程从队列中读取消息，并且为2个监听器分配了1个线程。

@Configuration
public class SqsListenerConfiguration {

    @Bean
    @ConfigurationProperties(prefix = "aws.configuration")
    public ClientConfiguration clientConfiguration() {
        return new ClientConfiguration();
    }


    @Bean
    @Primary
    public AWSCredentialsProvider awsCredentialsProvider() {

        ProfileCredentialsProvider credentialsProvider = new ProfileCredentialsProvider("credential");
        try {
            credentialsProvider.getCredentials();
            System.out.println(credentialsProvider.getCredentials().getAWSAccessKeyId());
            System.out.println(credentialsProvider.getCredentials().getAWSSecretKey());

        } catch (Exception e) {
            throw new AmazonClientException(
                    "Cannot load the credentials from the credential profiles file. " +
                            "Please make sure that your credentials file is at the correct " +
                            "location (~/.aws/credentials), and is in valid format.",
                    e);
        }
        return credentialsProvider;
    }


    @Bean
    @Primary
    public AmazonSQSAsync amazonSQSAsync() {
        return AmazonSQSAsyncClientBuilder.standard().
                withCredentials(awsCredentialsProvider()).
                withClientConfiguration(clientConfiguration()).
                build();
    }


    @Bean
    @ConfigurationProperties(prefix = "aws.queue")
    public SimpleMessageListenerContainer simpleMessageListenerContainer(AmazonSQSAsync amazonSQSAsync) {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
        simpleMessageListenerContainer.setAmazonSqs(amazonSQSAsync);
        simpleMessageListenerContainer.setMessageHandler(queueMessageHandler());
        simpleMessageListenerContainer.setMaxNumberOfMessages(10);
        simpleMessageListenerContainer.setTaskExecutor(threadPoolTaskExecutor());
        return simpleMessageListenerContainer;
    }


    @Bean
    public QueueMessageHandler queueMessageHandler() {
        QueueMessageHandlerFactory queueMessageHandlerFactory = new QueueMessageHandlerFactory();
        queueMessageHandlerFactory.setAmazonSqs(amazonSQSAsync());
        QueueMessageHandler queueMessageHandler = queueMessageHandlerFactory.createQueueMessageHandler();
        return queueMessageHandler;
    }


    @Bean
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(1);
        executor.setMaxPoolSize(1);
        executor.setThreadNamePrefix("oaoQueueExecutor");
        executor.initialize();
        return executor;
    }


    @Bean
    public QueueMessagingTemplate messagingTemplate(@Autowired AmazonSQSAsync amazonSQSAsync) {
        return new QueueMessagingTemplate(amazonSQSAsync);
    }


}

监听器配置

    @SqsListener(value = "${oao.sqs.url}", deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
    public void onMessage(String serviceData, @Header("MessageId") String messageId, @Header("ApproximateFirstReceiveTimestamp") String approximateFirstReceiveTimestamp) {

        System.out.println(" Data = " + serviceData + " MessageId = " + messageId);

        repository.execute(serviceData);
}

最佳答案

通过将corePoolSize和maximumPoolSize设置为相同，可以创建fixed-size thread pool。规则的很好解释已记录在here中

隐式设置maxPoolSize允许删除任务。
但是，默认队列容量是Integer.MAX_VALUE，实际上，它是无穷大。

需要注意的是，ThreadPoolTaskExecutor在下面使用了ThreadPoolExecutor，这在the docs中描述了一种不太常见的排队方法:



这意味着maxPoolSize仅在队列已满时才相关，否则线程数将永远不会超过corePoolSize。
例如，如果我们将从未完成的任务提交给线程池:

第一个corePoolSize提交将分别启动一个新线程；

之后，所有提交都进入队列；

如果队列是有限的并且其容量已用尽，则每个提交都会启动一个新线程，直到maxPoolSize为止；

当池和队列都已满时，新的提交将被拒绝。

排队-阅读docs

任何BlockingQueue均可用于传输和保留提交的任务。此队列的使用与池大小交互:

如果正在运行的核心线程数少于corePoolSize，则执行器始终
宁愿添加一个新线程而不是排队。

如果正在运行corePoolSize或更多线程，则执行器始终
更喜欢对请求进行排队，而不是添加新线程。

如果无法将请求排队，则除非创建新线程，否则
这将超过maximumPoolSize，在这种情况下，任务将
被拒绝。

如果线程数少于corePoolSize，请创建一个新的
用于运行新任务的线程。

如果线程数等于(或大于)corePoolSize，将任务放入队列。

如果队列已满，并且线程数少于maxPoolSize，创建一个新线程以在其中运行任务。

如果队列已满，并且线程数大于或
等于maxPoolSize，拒绝任务。