聚合計劃

Question

我想了解更多關於Apache 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添加到路由中，因爲在運行此程序時由於某種原因，主體爲空。

此外，如果你是一個初學者，想知道。我正在引入Maven的依賴關係，而不是將camel.jars放在我的Java類路徑中。

感謝您閱讀這個每個人。 ：D

那麼我該如何開始聚合消息Stackoverflow？ 0_o

和平，

user_loser

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; 
    } 
} 

然後在您的路線中引用它;所以它變成這樣的：

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.