Spring AMQP и Elasticsearch — объединение сообщений

Ответ на вопрос о потребителе и агрегаторе:

Конфигурация для приема сообщений от AMQP и агрегирования. Совокупная стратегия основана на фиксации транзакции:

<amqp:inbound-channel-adapter queue-names="myQueue"
                                  transaction-manager="transactionManager"
                                  channel-transacted="true"
                                  channel="aggregateChannel"
                                  advice-chain="aggregatorReaperAdvice"
                                  concurrent-consumers="4"
                                  tx-size="100"/>

<aggregator input-channel="aggregateChannel" output-channel="storeChannel"
                expire-groups-upon-completion="true"
                correlation-strategy-expression="T(Thread).currentThread().id"
                release-strategy-expression="^[payload.equals(@AGGREGATOR_RELEASE_MARK)] != null"
                expression="?[!payload.equals(@AGGREGATOR_RELEASE_MARK)].![payload]"/>

ReaperAdvice (код Groovy):

@Service
class AggregatorReaperAdvice implements MethodBeforeAdvice, InitializingBean {

    private static final TRANSACTION_RESOURCE_MARK = 'TRANSACTION_RESOURCE_MARK'

    public static final AGGREGATOR_RELEASE_MARK = 'AGGREGATOR_RELEASE_MARK'

    MessagingTemplate messagingTemplate

    @Autowired
    MessageChannel aggregateChannel

    @Override
    void afterPropertiesSet() throws Exception {
        Assert.notNull aggregateChannel, "aggregateChannel must not be null"
        messagingTemplate = new MessagingTemplate(aggregateChannel)
    }

    @Override
    void before(Method method, Object[] args, Object target) {
        if (!TransactionSynchronizationManager.hasResource(AggregatorReaperAdvice)) {
            TransactionSynchronizationManager.bindResource(AggregatorReaperAdvice, TRANSACTION_RESOURCE_MARK)
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {

                @Override
                void beforeCommit(boolean readOnly) {
                    messagingTemplate.send(MessageBuilder.withPayload(AGGREGATOR_RELEASE_MARK).build())
                }

                @Override
                void afterCompletion(int status) {
                    TransactionSynchronizationManager.unbindResource(AggregatorReaperAdvice)
                }

            })
        }
    }
}

Дайте мне знать, если это не ясно.

Все остальные вопросы будут рассмотрены в ближайшее время.

Для manual ack можно использовать channel.basicAck(deliveryTag, true); - для ack последнего deliveryTag для всех предыдущих сообщений.

Для случая headers["reply_to"]... Я думаю, вы должны предоставить пользовательский AbstractAggregatingMessageGroupProcessor для aggregator и убить двух зайцев: совокупный результат агрегатора и итерацию по MessageGroup.getMessages(), чтобы отправить каждого из них для процесса ответа предоставленному MessageChannel. Это быстрое решение для вашего случая.

Нечто похожее, но более слабо связанное решение может быть основано на результатах агрегатора и его MessageGroupStore, где вы извлекаете correlationKey для получения группы и ее сообщений для выполнения желаемой логики reply. В этом случае удалять группу из хранилища следует не агрегатором, а вручную после извлечения этой группы.