Как обрабатывать более 10 одновременных сообщений из очереди AWS SQS FiFo с помощью Spring Integration

Я хочу иметь возможность обрабатывать более 10 сообщений SQS одновременно с помощью Spring Integration Workflow.

Исходя из этого вопроса, было рекомендовано использовать ExecutorChannel. Я обновил свой код, но симптомы остались прежними.

Как выполнить поток интеграции Spring в несколько потоков для параллельного использования большего количества сообщений очереди Amazon SQS?

После этого обновления мое приложение запрашивает 10 сообщений, обрабатывает их, и только после того, как я вызову amazonSQSClient.deleteMessage ближе к концу потока, оно примет еще 10 сообщений из очереди SQS.

Приложение использует очередь SQS FiFo.

Есть ли что-то еще, что мне не хватает, или это неизбежный симптом использования SqsMessageDeletionPolicy.NEVER с последующим удалением сообщений в конце потока? Принятие сообщений в начале потока на самом деле не вариант из-за других ограничений.

Вот соответствующие фрагменты кода с некоторыми упрощениями, но я надеюсь, что он выражает проблему.

Конфигурация очереди

@Bean
public AsyncTaskExecutor inputChannelTaskExecutor() {
    SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
    executor.setConcurrencyLimit(50);
    return executor;
}

@Bean
@Qualifier("inputChannel")
public ExecutorChannel inputChannel() {
    return new ExecutorChannel(inputChannelTaskExecutor());
}

Я также попробовал ThreadPoolTaskExecutor вместо SimpleAsyncTaskExecutor с тем же результатом, но я включу и его, если он предлагает другое понимание.

    @Bean
    public AsyncTaskExecutor inputChannelTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("spring-async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.afterPropertiesSet();
        executor.initialize();
        return executor;
    }

Адаптер канала SQS

@Bean
public SqsMessageDrivenChannelAdapter changeQueueMessageAdapter() {
    SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(this.amazonSQSClient, changeQueue);
    adapter.setOutputChannel(inputChannel);
    adapter.setMessageDeletionPolicy(SqsMessageDeletionPolicy.NEVER);
    return adapter;
}


@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerSpec poller() {
    return Pollers.fixedRate(500, TimeUnit.MILLISECONDS).maxMessagesPerPoll(10);
}

Упрощенный основной поток

Обычный сценарий для нас - получить несколько правок Branch за короткий период времени. Этот поток "заботится" только о том, чтобы произошло хотя бы одно изменение. messageTransformer извлекает идентификатор из документа полезной нагрузки и помещает его в заголовок dsp_docId, который мы затем используем для агрегирования (мы используем этот идентификатор в нескольких других местах, поэтому мы почувствовал, что заголовок имеет смысл, а не выполняет всю работу в настраиваемом агрегаторе).

ProvisioningServiceActivator получает последнюю версию Branch, затем маршрутизатор решает, нужны ли ему дальнейшие преобразования (в этом случае он отправляет его в transformBranchChannel) или его можно отправить на наш экземпляр PI (через sendToPiChannel).

Поток преобразования (не показан, я не думаю, что он вам нужен) в конечном итоге приводит к отправке в поток PI, просто сначала он выполняет больше работы.

istingGroupProcessor захватывает все заголовки aws_receiptHandle и добавляет их в новый заголовок как | разделенный список.

Поток sendToPi (и errorFlow) заканчивается вызовом настраиваемого обработчика, который заботится об удалении всех сообщений SQS, на которые ссылается этот список строк aws_receiptHandle.

@Bean
IntegrationFlow sqsListener() {
    return IntegrationFlows.from(inputChannel)
                           .transform(messageTransformer)
                           .aggregate(a -> a.correlationExpression("1")
                                            .outputProcessor(listingGroupProcessor)
                                            .autoStartup(true)
                                            .correlationStrategy(message -> message.getHeaders().get("dsp_docId"))
                                            .groupTimeout(messageAggregateTimeout)  // currently 25s
                                            .expireGroupsUponCompletion(true)
                                            .sendPartialResultOnExpiry(true)
                                            .get())

                           .handle(provisioningServiceActivator, "handleStandard")
                           .route(Branch.class, branch -> (branch.isSuppressed() == null || !branch.isSuppressed()),
                                  routerSpec -> routerSpec.channelMapping(true, "transformBranchChannel")
                                                          .resolutionRequired(false)
                                                          .defaultOutputToParentFlow())

                           .channel(sendtoPiChannel)
                           .get();
}