差異處

這裏顯示兩個版本的差異處。

--- java:apache_camel:aggregator:helloworld [2019/03/17 21:57]
tony
+++ java:apache_camel:aggregator:helloworld [2023/06/25 09:48]
@@ 行 1: / 行 1: @@
-{{tag>camel}}
-====== Camel - Aggregator Hello World ======
-===== Introduction =====
-Camel Aggregator提供開發人員能夠將大量訊息合成一個的功能。假設你的系統會接受外來通知，並將內容寫進資料庫中；一次連線寫一筆資料會比較快，還是透過一次連線寫入五筆資料快呢? 正常來說，一次連線寫入五筆同屬性資料會是比較快的。它與[[java:apache_camel:throttler:helloworld|Throttler]]的最大區別在於，Throttler限制了client的請求，而Aggreator則是收集起來一次請求。因此，如果使用Aggreator，你的程式必須要有對應的處理方式。\\
-\\
-我將透過HTTP GET請求/events/{id}做為範例，說明如何使用Aggregator去將單位時間內的請求，以{id}去分組並Aggredate成各別分組訊息。本篇文章中使用到兩個RouteBuilder，分別為RestRouteBuilder與AggregatorGroupRouteBuilder。RestRouteBuilder負責REST核心相關設定，可參考[[java:apache_camel:throttler:helloworld|先前文章]]。接下來直接說明AggregatorGroupRouteBuilder。
-===== AggregatorGroupRouteBuilder =====
-<code java>
-package org.tonylin.practice.camel.aggregator;
-import static com.google.common.base.Preconditions.checkState;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy;
-public class AggregatorGroupRouteBuilder extends RouteBuilder {
-	private final static String GET_EVENTS = "GET_EVENTS";
-	private Object eventHandler;
-	private int period = 500;
-	public AggregatorGroupRouteBuilder(Object eventHandler) {
-		this.eventHandler = eventHandler;
-	}
-	public void setPeriod(int period) {
-		this.period = period;
-	}
-	@Override
-	public void configure() throws Exception {
-		checkState(eventHandler!=null, "Can't find eventHandler");
-		rest("/events/{id}").get().route().id(GET_EVENTS)
-		.aggregate(new GroupedExchangeAggregationStrategy())
-		.header("id")
-		.completionInterval(period)
-		.bean(eventHandler).endRest();
-	}
-}
-</code>
-以下是在configure中的幾個重點:
-  * aggregate(new GroupedExchangeAggregationStrategy()): 使用GroupedExchangeAggregationStrategy去做aggregate，aggregate的內容為Exchange。如果有任何條件限制，要以Exchange去操作。
-  * header("id"): 與aggregate一起使用，代表以header id去分組，在這裡就是屬於event的id。
-  * completionInterval(period): 做aggregate的單位時間。
-  * bean(eventHandler): 請求的處理者，必須要有能力處理aggregate後的訊息。
-接下來讓我們透過單元測試展示效果。
-===== Unit Test =====
-測試有兩個目標，testGroupedExchange用以確認aggregate group後的結果，另外一個testCompletionInterval則是確認單位時間設定的作用。以下為程式碼主要結構，主要測試程式碼稍後做說明:
-<code java>
-package org.tonylin.practice.camel.aggregator;
-import static com.google.common.base.Preconditions.checkState;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import org.apache.camel.Exchange;
-import org.apache.camel.Handler;
-import org.apache.camel.RoutesBuilder;
-import org.apache.camel.test.junit4.CamelTestSupport;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.junit.Test;
-import org.tonylin.practice.camel.rest.RestRouteBuilder;
-public class AggregatorGroupRouteBuilderTest extends CamelTestSupport  {
-	private RestHandler hander = new RestHandler();
-	private HttpClient client = HttpClientBuilder.create().build();
-	@Override
-	protected RoutesBuilder[] createRouteBuilders() throws Exception {
-		AggregatorGroupRouteBuilder aggregatorGroupRouteBuilder = new AggregatorGroupRouteBuilder(hander);
-		aggregatorGroupRouteBuilder.setPeriod(500);
-		return new RoutesBuilder[] {
-				new RestRouteBuilder(),
-				aggregatorGroupRouteBuilder
-		};
-	}
-	@Test
-	public void testGroupedExchange() throws Exception {
-		// skip
-	}
-	@Test
-	public void testCompletionInterval() throws Exception {
-		// skip
-	}
-}
-</code>
-此測試中的RestHandler程式碼如下，它會透過GROUPED_EXCHANGE取得aggregator處理後的內容，並根據header id儲存到map中，供測試程式碼做assertion。除此之外，在這裡使用CountDownLatch了去確認有收到預期訊息數量:
-<code java>
-public static class RestHandler {
-	private Map<String, List<Exchange>> requestData = new ConcurrentHashMap<String, List<Exchange>>();
-	private CountDownLatch countDownLatch;
-	@SuppressWarnings("unchecked")
-	@Handler
-	public void handle(Exchange exchange) {
-		List<Exchange> groupExchanges = exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
-		String id = groupExchanges.get(0).getIn().getHeader("id", String.class);
-		requestData.put(id, groupExchanges);
-		if( countDownLatch != null )
-			countDownLatch.countDown();
-	}
-	public Map<String, List<Exchange>> getRequestData(){
-		return requestData;
-	}
-	public void expectNum(int num) {
-		countDownLatch = new CountDownLatch(num);
-	}
-	public void waitCompletion(int timeout) throws InterruptedException {
-		checkState(countDownLatch!=null);
-		countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
-	}
-	public void clear() {
-		requestData.clear();
-	}
-}
-</code>
-首先讓我說明testGroupedExchange。測試程式碼中，送了四個訊息，其中包含了三組不同id；這樣的請求，我們預期RestHandler中，總共會收到三組訊息，其中id 123那組，應包含兩個訊息:
-<code java>
-private HttpResponse requestWithEventId(String id) {
-	try {
-		HttpGet httpGet = new HttpGet("http://localhost:8080/events/" + id);
-		return client.execute(httpGet);
-	} catch( Exception e ) {
-		throw new RuntimeException(e);
-	}
-}
-@Test
-public void testGroupedExchange() throws Exception {
-	// given request event id
-	List<String> eventIds = Arrays.asList("123", "123", "456", "789");
-	hander.expectNum(3);
-	// when
-	List<HttpResponse> responses = eventIds.parallelStream().map(this::requestWithEventId).collect(Collectors.toList());
-	// then
-	responses.forEach(response->{
-		assertEquals(200, response.getStatusLine().getStatusCode());
-	});
-	hander.waitCompletion(1000);
-	Map<String, List<Exchange>> requestData = hander.getRequestData();
-	assertEquals(3, requestData.size());
-	assertEquals(2, requestData.get("123").size());
-	assertEquals(1, requestData.get("456").size());
-	assertEquals(1, requestData.get("789").size());
-}
-</code>
-接著是testCompletionInterval，這裡我直接透過請求相同id兩次，並分兩輪送出。這樣做可以確認aggregator有做到根據completionInterval的分批效果:
-<code java>
-private void requestTwiceWithId(String id) throws Exception {
-	// request event {id} twice
-	List<String> eventIds = Arrays.asList(id, id);
-	hander.expectNum(1);
-	List<HttpResponse> responses = eventIds.parallelStream().map(this::requestWithEventId).collect(Collectors.toList());
-	responses.forEach(response->{
-		assertEquals(200, response.getStatusLine().getStatusCode());
-	});
-	hander.waitCompletion(1000);
-	// confirm request result
-	Map<String, List<Exchange>> requestData = hander.getRequestData();
-	assertEquals(1, requestData.size());
-	assertEquals(2, requestData.get("123").size());
-	// clear first request data
-	hander.clear();
-	assertTrue(hander.getRequestData().isEmpty());
-}
-@Test
-public void testCompletionInterval() throws Exception {
-	requestTwiceWithId("123");
-	requestTwiceWithId("123");
-}
-</code>
-Camel Aggregator其實提供了很多細部操作，你也可以根據自身需求做AggregationStrategy；但時間有限，請容我以後有機會再分享。
-===== Library Info (Gradle Config) =====
-以下是我在寫這篇文章時，所使用的libraries版本:
-<code>
-ext {
-	camelVersion='2.23.1'
-	nettyAllVersion='4.1.34.Final'
-	guavaVersion='27.1-jre'
-	log4jVersion='1.2.17'
-	slf4jVersion='1.7.26'
-	httpClientVersion='4.5.7'
-}
-dependencies {
-    compile group: 'org.apache.camel', name: 'camel-core', version: "$camelVersion"
-    compile group: 'org.apache.camel', name: 'camel-netty4-http', version: "$camelVersion"
-    compile group: 'org.apache.camel', name: 'camel-http-common', version: "$camelVersion"
-    compile group: 'org.apache.camel', name: 'camel-netty4', version: "$camelVersion"
-    compile group: 'io.netty', name: 'netty-all', version: "$nettyAllVersion"
-    compile group: 'com.google.guava', name: 'guava', version: "$guavaVersion"
-    compile group: 'log4j', name: 'log4j', version: "$log4jVersion"
-    compile group: 'org.slf4j', name: 'slf4j-api', version: "$slf4jVersion"
-    runtime group: 'org.slf4j', name: 'slf4j-log4j12', version: "$slf4jVersion"
-    testCompile group: 'org.apache.camel', name: 'camel-test', version: "$camelVersion"
-    testCompile group: 'org.apache.httpcomponents', name: 'httpclient', version: "$httpClientVersion"
-    testCompile 'junit:junit:4.12'
-}
-</code>
-===== Reference =====
-  * [[http://camel.apache.org/aggregator.html|Camel - Aggregator]]
-=====    =====
-----
-\\
-~~DISQUS~~