如何通过spring integration Aggregator将一条消息添加到多个组中

Question

嗨，我们正在尝试流式处理金融市场数据，通过利用apache camel或Spring集成来计算交易信号。我们的一个用例是根据价格时间戳将连续价格汇总在一起，如下所示：

• 输入

输入消息以时间序列中的（时间戳、价格）对的形式出现。假设值为，每对（TX，PX）是一条消息，而T表示时间戳，P表示价格值

(T0,P1),(T1,P1),(T2,P2),(T3,P3),(T4,P4)... 

• 聚合

假设我们需要将每3条连续消息聚合在一起进行进一步计算，给定需要生成以下组的输入消息，每个3对组都是聚合消息：

[(T0,P1),(T1,P1),(T2,P2)],
[(T1,P1),(T2,P2),(T3,P3)],
[(T2,P2),(T3,P3),(T4,P4)],
....

正如您所看到的，大多数消息将聚合到多个组。是否有人可以建议是否有一种方法可以做到这一点，即使用当前的聚合器而不编写聚合器。

spring集成聚合分组似乎也基于相关键，因此消息需要映射到一组相关键。然而，当前的api似乎只允许我们生成一个关联键，这意味着每个消息只能聚合到一个组。这方面有什么解决办法吗。

附笔。

在阅读了camel的源代码之后，camel似乎无法支持我们的需求。试试Spring的运气吧。交叉手指骆驼问题

Answer 1

我们没有任何现成的东西，但我可以通过对SimpleMessageStore进行一个小小的修改来实现您的愿望。我已经在摘要中发布了完整的RollingMessageStore。

底线是修改removeGroup，只删除第一条消息，而不是整个组。此外，将completeGroup设置为no-op。

设置expreGroupOnComplection以强制聚合器“删除”组（通过调用修改后的RemoveGroup（）方法。

这是SimpleMessageGroup和RollingMessageGroup之间的区别...

182,184c190,194
< 
<               groupUpperBound.release(groupIdToMessageGroup.get(groupId).size());
<               groupIdToMessageGroup.remove(groupId);
---
>               Message<?> message = this.groupIdToMessageGroup.get(groupId).getOne();
>               if (message != null) {
>                   this.groupUpperBound.release(1);
>                   this.removeMessageFromGroup(groupId, message);
>               }

（加上删除完整组（）中的所有代码。

还有一个测试用例。。。

@Test
public void testRolling() {
    AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), new RollingMessageStore());
    aggregator.setExpireGroupsUponCompletion(true);
    aggregator.setReleaseStrategy(new ReleaseStrategy() {

        @Override
        public boolean canRelease(MessageGroup group) {
            return group.size() == 3;
        }
    });
    QueueChannel replyChannel = new QueueChannel();
    Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
    Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
    Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, null);
    Message<?> message4 = createMessage(9, "ABC", 3, 3, replyChannel, null);
    Message<?> message5 = createMessage(11, "ABC", 3, 3, replyChannel, null);

    aggregator.handleMessage(message1);
    aggregator.handleMessage(message2);
    aggregator.handleMessage(message3);
    aggregator.handleMessage(message4);
    aggregator.handleMessage(message5);

    Message<?> reply = replyChannel.receive(10000);
    assertNotNull(reply);
    assertEquals(reply.getPayload(), 105);
    reply = replyChannel.receive(10000);
    assertNotNull(reply);
    assertEquals(reply.getPayload(), 315);
    reply = replyChannel.receive(10000);
    assertNotNull(reply);
    assertEquals(reply.getPayload(), 693);
}

请继续并打开JIRA的新功能问题，我们将考虑将此（或更通用的解决方案）添加到即将发布的3.0版本中。

使用相关策略表达式=“foo”和

发布-策略-表达式=size（）==3。