SpringBoot @SqsListener - не работает - с исключением - TaskRejectedException

Проблема в конфигурации потока слушателя. См. Следующие

...
ThreadPoolExecutor@7c1594a5[Running, pool size = 3, active threads = 3, queued tasks = 0, completed tasks = 20]]
...

Размер пула потоков по умолчанию меньше желаемого.

Добавьте следующую конфигурацию в свое приложение Spring

@Configuration
public class TasksConfiguration implements SchedulingConfigurer {

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setPoolSize(5); // TODO: Load this from configuration
        taskScheduler.initialize();
        taskRegistrar.setTaskScheduler(taskScheduler);
    }
}

Теперь у вас должна быть возможность обрабатывать эти задачи.

P.S. Какие бы задачи не были отклонены ранее, они будут подняты позже, по прошествии определенного периода.

Обновлено: я думаю, что люди пугаются числа в строке .setPoolSize(5000). Это настраиваемый номер, который вы можете выбрать в соответствии с вашими требованиями. Для ответа я сокращаю его до меньшего числа.

установка максимального количества сообщений, похоже, решает проблему:

@Bean
public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(AmazonSQSAsync amazonSQS){
    SimpleMessageListenerContainerFactory factory = new SimpleMessageListenerContainerFactory();
    factory.setAmazonSqs(amazonSQS);
    factory.setMaxNumberOfMessages(10);
    return factory;
}

Привет, я решил эту проблему с помощью Spring Listener. Ниже приведен код, надеюсь, он поможет.

В следующем решении после завершения инициализации всех bean-компонентов выделяется новый исполнитель задачи с большим размером пула.

@Component
public class PostBeansConstructionListener{

    @EventListener
    public void handleContextRefreshedEvent(ContextRefreshedEvent event){
        final ApplicationContext applicationContext = event.getApplicationContext();
        final SimpleMessageListenerContainer simpleMessageListenerContainer = applicationContext.getBean(SimpleMessageListenerContainer.class);
        setAsyncTaskExecutor(simpleMessageListenerContainer);
    }

    private void setAsyncTaskExecutor(SimpleMessageListenerContainer simpleMessageListenerContainer) {
        try{
            simpleMessageListenerContainer.setTaskExecutor(getAsyncExecutor());
        }catch(Exception ex){
            throw new RuntimeException("Not able to create Async Task Executor for SimpleMessageListenerContainer.", ex);
        }
    }

    public AsyncTaskExecutor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(7);
        executor.setMaxPoolSize(42);
        executor.setQueueCapacity(11);
        executor.setThreadNamePrefix("threadPoolExecutor-SimpleMessageListenerContainer-");
        executor.initialize();
        return executor;
    }
}

Невозможно добавить комментарий к предыдущим ответам, чтобы дополнительно объяснить, почему возникает проблема и почему работает параметр решения MaxNumberOfMessages сообщений. Надеюсь, следующее поможет все прояснить.

SimpleMessageListenerContainerThreadPoolTaskExecutor настроен на размер основного пула 2 потока, максимальный размер пула 3 потока и емкость очереди 0. Однако по умолчанию максимальное количество сообщений, возвращаемых при опросе в Amazon SQS, равно 10. Значение что если в одном опросе будет доступно 10 сообщений, потоков для их обработки не хватит. Таким образом выбрасывается RejectedExecutionException.

Настройка setMaxNumberOfMessages на 10 на SimpleMessageListenerContainerFactory устанавливает максимальный размер пула потоков равным 11, что должно обеспечить доступность достаточного количества потоков. Он не устанавливает емкость очереди.

Чтобы установить емкость очереди, можно инициализировать отдельный TaskExecutor и установить его в компоненте SimpleMessageListenerContainerFactory следующим образом:

@Bean(name = "sqsAsyncTaskExecutor")
public AsyncTaskExecutor asyncTaskExecutor(@Value("${external.aws.sqs.core-thread-count}") int coreThreadCount,
                                           @Value("${external.aws.sqs.max-thread-count}") int maxThreadCount,
                                           @Value("${external.aws.sqs.queue-capacity}") int queueCapacity) {
    ThreadPoolTaskExecutor asyncTaskExecutor = new ThreadPoolTaskExecutor();
    asyncTaskExecutor.setCorePoolSize(coreThreadCount);
    asyncTaskExecutor.setMaxPoolSize(maxThreadCount);
    asyncTaskExecutor.setQueueCapacity(queueCapacity);
    asyncTaskExecutor.setThreadNamePrefix("threadPoolExecutor-SimpleMessageListenerContainer-");
    asyncTaskExecutor.initialize();
    return asyncTaskExecutor;
}

@Bean
public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(AmazonSQSAsync amazonSQS, @Qualifier("sqsAsyncTaskExecutor") AsyncTaskExecutor asyncTaskExecutor) {
    SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory = new SimpleMessageListenerContainerFactory();
    simpleMessageListenerContainerFactory.setTaskExecutor(asyncTaskExecutor);
    return simpleMessageListenerContainerFactory;
}

Я использовал следующие значения: coreThreadCount = 5, maxThreadCount = 20, queueCapacity = 10.

Как я уже сказал, я думаю, что настройки setMaxNumberOfMessages на 10 на SimpleMessageListenerContainerFactory должно быть достаточно для обработки всех пакетных сообщений, полученных из одного запроса. Однако, если вы чувствуете, что вам нужен более точный контроль над TaskExecutor, эта конфигурация также подойдет.

Я считаю, что весной это ошибка или недосмотр. Проблема связана со значениями по умолчанию:

public class SimpleMessageListenerContainer extends AbstractMessageListenerContainer {

    private static final int DEFAULT_WORKER_THREADS = 2;

а также

abstract class AbstractMessageListenerContainer implements InitializingBean, DisposableBean, SmartLifecycle, BeanNameAware {
    private static final int DEFAULT_MAX_NUMBER_OF_MESSAGES = 10;

Если maxNumberOfMessages не задан, то он использует 10 как количество сообщений, извлекаемых из SQS, и 2 как количество рабочих в исполнителе задачи. Это означает, что если он извлекает 3 или более сообщения одновременно, вы получаете это исключение. Если вы вручную установите maxNumberOfMessages на значение (любое значение), он будет использовать его в обоих местах, синхронизируя значения, как я считаю, ожидается:

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(
            SimpleMessageListenerContainerFactory factory, QueueMessageHandler messageHandler)
    {
        SimpleMessageListenerContainer container = factory.createSimpleMessageListenerContainer();
        container.setMaxNumberOfMessages(5);
        container.setMessageHandler(messageHandler);
        return container;
    }