当消息计数大于并发消费者的数量时,如何使用Spring IntegrationFlow中所需的所有消息?

我有一个定义如下的集成流:

IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "queueName")
                    .id("id")
                    .autoStartup(autoStartup)
                    .concurrentConsumers(2)
                    .maxConcurrentConsumers(3)
                    .messageConverter(messageConverter()))
                    .aggregate(a -> a.correlationExpression("payload.entityId")
                                    .releaseExpression("size() eq iterator().next().payload.batchSize")
                                    .sendPartialResultOnExpiry(true)
                                    .groupTimeout(2000)
                                    .expireGroupsUponCompletion(true)
                                    .outputProcessor(myMessageGroupProcessor))
                    .handle(serviceActivatorBean, "myMethod", e -> e.advice(requestHandlerRetryAdviceForIntegrationFlow()))
                    .get();

其目的是将“批量”发送的多条相关消息分组。举个例子:

// Message 1
{ "name": "Message1", 
  "entityId": "someId"
  "batchSize": 2,
  "batchIndex": 1, 
  .... }

// Message 2
{ "name": "Message2",
  "entityId": "someId"
  "batchSize": 2,
  "batchIndex": 2, 
  .... }

由于所述原因 here 我们正在使用RabbitMQ的手动确认,以避免丢失消息。

集成流对于大小为2的批处理效果很好,但一旦一个批处理中有2条以上的消息,我们就会遇到麻烦:

[my-service] 2017-12-04 17:46:07.966  INFO 1 --- [ask-scheduler-5] x.y.EntityUpdater : Will update entity [entitId] from messages: Message1, Message2 
[my-service] 2017-12-04 17:46:09.976  INFO 1 --- [ask-scheduler-3] x.y.EntityUpdater : Will update entity [entitId] from messages: Message3

请注意,记录的消息之间的时间大约为2秒(即我们所确认的时间) groupTimeout ).

我怀疑这是因为Spring消耗了2条消息(不是自动确认的),然后聚合等待第3条消息(因为 batchSize 在本例中为3)。但是,由于只有两个并发消费者,因此该消息永远不会在2秒钟的窗口内被消费。

增加 concurrentConsumers 计数到3个解 这 特定问题。问题是,我们不知道我们收到的批次的大小,它们可能相当大,可能大小在50左右。这意味着简单地增加 concurrentConsumers公司 这不是一个可行的选择。

春天处理这个问题的合适方法是什么?