Spring Integration：如何一次处理多条消息？

Question

我有以下配置：

<bean id="mongoDbMessageStore" class="org.springframework.integration.mongodb.store.MongoDbMessageStore">
    <constructor-arg ref="mongoDbFactoryDefault"/>
</bean>

<!-- the queue capacity is unbounded as it uses a persistent store-->
<int:channel id="logEntryChannel">
    <int:queue message-store="mongoDbMessageStore"/>
</int:channel>

<!-- the poller will process 10 messages every 6 seconds -->
<int:outbound-channel-adapter channel="logEntryChannel" ref="logEntryPostProcessorReceiver" method="handleMessage">
    <int:poller max-messages-per-poll="10" fixed-rate="6000"/>
</int:outbound-channel-adapter>

而消息处理程序定义为

@Override
public void handleMessage(Message<?> message) throws MessagingException {
    Object payload = message.getPayload();
    if (payload instanceof LogEntry) {
        LogEntry logEntry = (LogEntry) payload;
        String app = (String) message.getHeaders().get("app");
        logger.info("LogEntry Received - " + app + " " + logEntry.getEntityType() + " " + logEntry.getAction() + " " + logEntry.getEventTime());
        logEntryPostProcessService.postProcess(app, logEntry);
    } else {
        throw new MessageRejectedException(message, "Unknown data type has been received.");
    }
}

我想要的是类似的东西

@Override
public void handleMessage(List<Message<?>> messages) throws MessagingException {
...
}

所以基本上轮询器在一次调用中发送所有 10 条消息，而不是每条消息调用该方法 10 次。

这样做的原因是有可能批量处理块中的所有消息，从而提高性能。

Answer 1

感谢@Artem Bilan，这里是最终解决方案：

<bean id="mongoDbMessageStore" class="org.springframework.integration.mongodb.store.MongoDbMessageStore">
    <constructor-arg ref="mongoDbFactoryDefault"/>
</bean>

<!-- the queue capacity is unbounded as it uses a persistent store-->
<int:channel id="logEntryChannel">
    <int:queue message-store="mongoDbMessageStore"/>
</int:channel>

<!-- 
    the poller will process 100 messages every minute 
    if the size of the group is 100 (the poll reached the max messages) or 60 seconds time out (poll has less than 100 messages) then the payload with the list of messages is passed to defined output channel
-->
<int:aggregator input-channel="logEntryChannel" output-channel="logEntryAggrChannel"
    send-partial-result-on-expiry="true"
    group-timeout="60000"
    correlation-strategy-expression="T(Thread).currentThread().id"
    release-strategy-expression="size() == 100">
    <int:poller max-messages-per-poll="100" fixed-rate="60000"/>
</int:aggregator>

<int:channel id="logEntryAggrChannel"/>        

<!-- the payload is a list of log entries as result of the aggregator -->
<int:outbound-channel-adapter channel="logEntryAggrChannel" ref="logEntryPostProcessorReceiver" method="handleMessage"/>

根据评论（在上面的代码中），我必须设置 group-timeout/send-partial-result-on-expiry，因为某些组是使用 Thread ID 形成的，但从未处理过，因为它们没有达到条件大小 == 100。

Answer 2

确实如此，因为 (AbstractPollingEndpoint)：

taskExecutor.execute(new Runnable() {
    @Override
    public void run() {
        int count = 0;
        while (initialized && (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll)) {
...
            if (!pollingTask.call()) {
                break;
            }
...
    }
});

因此，您的所有消息 (max-messages-per-poll) 都在同一个线程中处理。 但是，它们会被一个一个地发送给处理程序，而不是作为一个整体发送给处理程序。

要并行处理，您应该在 logEntryPostProcessorReceiver 之前使用 ExecutorChannel。像这样的：

<channel id="executorChannel">
   <dispatcher task-executor="threadPoolExecutor"/>
</channel>

<bridge input-channel="logEntryChannel" output-channel="executorChannel">
   <poller max-messages-per-poll="10" fixed-rate="6000"/>
</bridge>

<outbound-channel-adapter channel="executorChannel" ref="logEntryPostProcessorReceiver" method="handleMessage"/>

更新

要将消息作为一批处理，您应该aggregate 他们。由于它们都是polling endpoint 的结果，因此消息中没有sequenceDetails。你可以用correlationId 的一些假值来克服它：

<aggregator correlation-strategy-expression="T(Thread).currentThread().id"
        release-strategy-expression="size() == 10"/>

其中size() == 10 应该等于max-messages-per-poll。

之后您的logEntryPostProcessorReceiver 必须应用payloads 的list。或者只是一条消息，其中payload 是<aggregator> 的结果列表。