java - 聚合计划

Question

我正在尝试了解有关 Apache Camel 的更多信息。我发现该文档有些帮助，但给 Camel 初学者留下了很多猜测，他们不知道小代码段应该如何轻松地融入到功能齐全的程序中。希望大多数人都知道我想描述什么。我在一些编程书籍中迷失了很多次，其中的代码段显示在完全运行的程序的上下文之外。

无论如何，这是我的程序，由于某种原因不会聚合消息。我希望它能汇总我所有的消息，但这个程序并没有这样做。运行程序时，我收到一个空文件作为输出，这不是我的目标。

package laser.helmet.camel.friend;

import org.apache.camel.builder.RouteBuilder;

public class AggregatingMessagesRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        from("direct:start")
                .aggregate().constant(true)
                .completionTimeout(100L)
                .groupExchanges()
                .to("file:target/this_folder/result?allowNullBody=true");
    }

}

然后我从下面的类中调用这个程序，当然它有 main 方法。

package laser.helmet.camel.friend;

import org.apache.camel.CamelContext;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.ProducerTemplate;

public class Main {

    public static void main(String[] args) throws Exception { 
        CamelContext c = new DefaultCamelContext();
        c.addRoutes(new AggregatingMessagesRoute());
        ProducerTemplate pt = c.createProducerTemplate();

        c.start();

        pt.sendBody("direct:start", "1");
        pt.sendBody("direct:start", "2");
        Thread.sleep(5000);
        c.stop();
    }

}

我希望在路由完成后，使用 ProducerTemplate 创建的两条消息的正文都位于文件中，但它只是一个空白文件。我必须将 allowNullBody=true 部分添加到 route ，因为由于某种原因，运行该程序时主体为空。

如果您是初学者并且想知道。我将使用 Maven 引入依赖项，而不是将camel.jar 放在我的 Java 类路径中。

感谢大家阅读本文。 :D

那么我如何开始聚合 Stackoverflow 消息呢？ 0_o

和平，

用户失败者

Answer 1

这是一个简单的事情，因为您提供了代码片段，很好!删除 .groupExchanges() 因为它已被弃用并且它几乎不够用。您始终需要 AggregationStrategy 来对如何聚合交换进行细粒度控制。因此，添加以下类:

class StringAggregationStrategy 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(oldBody + "+" + newBody);
        return oldExchange;
    }
}

然后在你的 route 引用它；所以它变成了这样:

public class AggregatingMessagesRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        from("direct:start")
                .aggregate().constant(true)
                .completionTimeout(100L)
                .aggregationStrategy(new StringAggregationStrategy())
                .to("file:target/this_folder/result?allowNullBody=true");
    }

}

希望这有帮助!

R。