java - 如何使用驼峰聚合将一条消息聚合到多个组中？

Question

我正在尝试生成连续市场数据的聚合 View ，这意味着我们需要计算每 2 条消息的总和值。假设传入的数据为:

(V0,T0),(V1,T1),(V2,T2),(V3,T3)....

V 表示值 T 表示我们收到数据时的时间戳。

我们需要为每 2 个点生成总和:

(R1=Sum(V0,V1),T1),(R2=Sum(V1,V2),T2),(R3=Sum(V2,V3),T3),....

有什么建议我们如何使用aggregator2来做到这一点，或者我们需要为此编写一个处理器？

Answer 1

你说得对，aggregator2组件是一个好的方法。我会尝试这样的事情:

from("somewhere").split(body().tokenize("),")).streaming()
    .aggregate(new ValueAggregationStrategy()).completionTimeout(1500)
    .to("whatYouWant");

class ValueAggregationStrategy implements AggregationStrategy {

    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        if (oldExchange == null) {
            return newExchange;
        }

        String oldBody = oldExchange.getIn().getBody(String.class);
        String newBody = newExchange.getIn().getBody(String.class);

        oldExchange.getIn().setBody(extractValue(oldBody) + extractValue(newBody));
        return oldExchange;
    }

    public int extractValue(String body) {
        // Do the work "(V0,T0" -> "V0"
    }
}

注意:如果你可以有这样的格式，解析会更容易:V0,T0;V1,T1...

了解更多信息: here是Claus Ibsen写的一篇关于用Camel解析大文件的文章