架構設計：系統間通訊（39）——Apache Camel快速入門（下2）
4-2-1、LifecycleStrategy
LifecycleStrategy介面按照字面的理解是一個關於Camel中元素生命週期的規則管理器，但實際上LifecycleStrategy介面的定義更確切的應該被描述成一個監聽器：
當Camel引用程式中發生諸如Route載入、Route移除、Service載入、Serivce移除、Context啟動或者Context移除等事件時，DefaultCamelContext中已經被新增到集合“lifecycleStrategies”（java.util.List<LifecycleStrategy>）的LifecycleStrategy物件將會做相應的事件觸發。
讀者還應該注意到“lifecycleStrategies”集合是一個CopyOnWriteArrayList，我們隨後對這個List的實現進行講解。以下程式碼展示了在DefaultCamelContext新增Service時，DefaultCamelContext內部是如何觸發“lifecycleStrategies”集合中已新增的監聽的：
......
private void doAddService(Object object, boolean closeOnShutdown) throws Exception {
......
// 只有以下條件成立，才需要將外部來源的Object作為一個Service處理 4-2-2、CopyOnWriteArrayList與監聽者模式
正如上一小節講到的，已在DefaultCamelContext中註冊的LifecycleStrategy物件存放於一個名叫“lifecycleStrategies”的集合中，後者是CopyOnWriteArrayList容器的實現，這是一個從JDK 1.5+ 版本開始提供的容器結構。
各位讀者可以設想一下這樣的操作：某個執行緒在對容器進行寫操作的同時，還有另外的執行緒對容器進行讀取操作。如果上述操作過程是在沒有“執行緒安全”特性的容器中進行的，那麼可能出現的情況就是：開發人員原本想讀取容器中 i 位置的元素X，可這個元素已經被其它執行緒刪除了，開發人員最後讀取的 i 位置的元素變成了Y。但是在具有“寫執行緒安全”特性的容器中進行這樣的操作就不會有問題：因為寫操作在另一個副本容器中進行，原容器中的資料大小、資料位置都不會受到影響。
如果上述操作過程是在有“執行緒安全”特性的容器中進行的，那麼以上髒讀的情況是可以避免的。但是又會出現另外一個問題：由於容器的各種讀寫操作都會加上鎖（無論是悲觀鎖還是樂觀鎖），所以容器的讀寫效能又會收到影響。如果採用的是樂觀鎖，那麼對效能的影響可能還不會太大，但是如果採用的是悲觀鎖，那麼對效能的影響就有點具體了。
CopyOnWriteArrayList為我們提供了另一種執行緒安全的容器操作方式。CopyOnWriteArrayList的工作效果類似於java.util.ArrayList，但是它通過ReentrantLock實現了容器中寫操作的執行緒安全性。CopyOnWriteArrayList最大的特點是：當進行容器中元素的修改操作時，它會首先將容器中的原有元素克隆到一個副本容器中，然後對副本容器中的元素進行修改操作。待這些操作完成後，再將副本中的元素集合重新會寫到原有的容器中完成整個修改操作。這種工作機制稱為Copy-On-Write（COW）。這樣做的最主要目的是分離容器的讀寫操作。CopyOnWriteArrayList會對所有的寫操作加鎖，但是不會對任何容器的讀操作加鎖（因為寫操作在一個副本中進行）。
另外CopyOnWriteArrayList還重新實現了一個新的迭代器：COWIterator。它是做什麼的呢？舉例說明：在ArrayList中我們如果在進行迭代時同時進行容器的寫操作，那麼就可能會因為下標超界等原因出現程式異常：
List<?> list = new ArrayList<?>();
// 省略了新增元素部分的程式碼
......
// ArrayList不支援這樣的操作方式，會報錯
for(Object item : list){
list.remove(item);
}但如果使用CopyOnWriteArrayList中重寫的COWIterator迭代器，就不會出現的情況（開發人員還可以使用JDK 1.5+ 提供的另一個執行緒安全COW容器：CopyOnWriteArraySet）：
List<?> list = new CopyOnWriteArrayList<?>();
// 省略了新增元素部分的程式碼
......
// COWIterator迭代器支援一邊迭代一邊進行容器的寫操作
for(Object item : list){
list.remove(item);
}那麼CopyOnWriteArrayList和監聽器模式有什麼關係呢？在書本上我們學到的監聽器容器基本上都不是執行緒安全的，這基本上是出於兩方面的考慮。首先對於設計模式的初學者來說最重要的理解模式所代表的設計思想，而非實現細節；另外，在這些示例中，設計模式的實現和操作一般為單一執行緒，不會出現多其它執行緒同時操作容器的情況。以下是我們常看到的監聽者模式（程式碼片段）：
/**
* 為事件監聽攜帶的業務物件
* @author yinwenjie
*/
public class BusinessEventObject extends EventObject {
public BusinessEventObject(Object source) {
super(source);
}
}
/**
* 監聽器，其中只有一個事件方法
* @author yinwenjie
*/
public interface BusinessEventListener extends EventListener {
public void onBusinessStart(BusinessEventObject eventObject);
}
/**
* 業務級別的程式碼
* @author yinwenjie
*/
public class BusinessOperation {
/**
* 已註冊的監聽器放在這裡
*/
private List<BusinessEventListener> listeners = new ArrayList<BusinessEventListener>();
public void registeListener(BusinessEventListener eventListener) {
this.listeners.add(eventListener);
}
......
public void doOp() {
//業務程式碼在這裡執行後，接著促發監聽
for (BusinessEventListener businessEventListener : listeners) {
businessEventListener.onBusinessStart(new BusinessEventObject(this));
}
}
......
}以上程式碼無需做太多說明。請注意，由於我們使用ArrayList這樣的非執行緒安全容器作為已註冊監聽的儲存容器，所以開發人員在使用這個容器觸發監聽事件時需要格外小心：確保同一時間只會有一個執行緒對容器進行寫操作、確保在一個迭代器內沒有容器的寫操作、還要確保每個監聽器的具體實現不會把當前執行緒鎖死（次要）——但作為開發人員真的能隨時保證這些事情嗎？
4-2-3、SoftReference
我們都知道JVM的記憶體是有上限的，JVM的垃圾回收執行緒進行工作時會將當前沒有任何引用可達性的物件區域進行回收，以便保證JVM的記憶體空間能夠被迴圈利用。當JVM的可用記憶體達到上限，且垃圾回收執行緒又無法找到任何可以回收的物件時，應用程式就會報錯。JVM中某個執行緒的堆疊狀態可能如下圖所示：
（圖A）
上圖中執行緒Thread1在執行時，在棧記憶體中建立了一個變數X。變數X指向堆記憶體為A類例項化物件分配的記憶體空間（後文稱之為A物件）。注意，A物件中還對同樣存在於堆記憶體區域中的B類、C類的例項化物件（後文稱為B物件、C物件）有引用關係。那麼如果JVM垃圾回收策略要對A物件、B物件、C物件三個記憶體區域進行回收，除非針對這些區域的引用可達性全部消失，否則以上所說到的對記憶體區域都不會被回收。這樣的物件間引用方式被稱為強引用（Strong Reference）：JVM寧願丟擲OutOfMemoryError也不會在還存在引用可及性的情況下回收記憶體區域。
引用可達性，是JVM垃圾回收策略中確認哪些記憶體區域可以進行回收的判斷演算法。大致的定義是：從某個根引用開始進行引用圖結構的深度遍歷掃描，當遍歷完成時那些沒有被掃描到的一個（或者多個）記憶體區域就是失去引用可達性的區域。
JAVA JDK1.2+開始還提供一種稱為“軟引用”（Soft Reference）的物件間引用方式。在這種方式下，物件間的引用關係通過一個命名為java.lang.ref.SoftReference的工作類進行間接託管，目的是當JVM記憶體空間不足，垃圾回收策略被主動觸發時 進行以下回收策略操作：掃面當前堆記憶體中只建立了“軟引用”的記憶體區域，無論這些“軟引用”是否依然存在引用可達性，都強制對這些建立了“軟引用”的物件進行回收，以便騰出記憶體空間。下面我們對圖A中的物件間引用關係進行如下圖所示的調整：
上如所示的引用關係和圖A中的引用關係類似，只是我們在A對B、C的引用關係上都增加了一個SoftReference物件進行間接關聯。程式碼片段如下所示：
package com.test;
import java.lang.ref.SoftReference;
public class A {
/**
* 軟引用 B
*/
private SoftReference<B> paramB;
/**
* 軟引用 C
*/
private SoftReference<C> paramC;
/**
* 建構函式中，建立和B、C的軟引用
* @param paramB
* @param paramC
*/
public A(B paramB , C paramC) {
this.paramB = new SoftReference<B>(paramB);
this.paramC = new SoftReference<C>(paramC);
}
/**
* @return the paramB
*/
public B getParamB() {
return paramB.get();
}
/**
* @return the paramC
*/
public C getParamC() {
return paramC.get();
}
}當出現“軟引用”物件被垃圾回收執行緒回收時，例如B物件被回收時，A物件中的getB()方法將會返回null。那麼原來進行B物件間接引用動作的SoftReference物件該怎麼處理呢？要知道如果B物件被回收了，那麼承載這個“軟引用”的SoftReference物件就沒有什麼用處了。還好JDK中幫我們準備了名叫ReferenceQueue的佇列，當SoftReference物件所承載的“軟引用”物件被回收後，這個Reference物件將被送入ReferenceQueue中（當然你也可以不指定，如果不指定的話SoftReference物件會以“強引用”的回收策略被回收，不過SoftReference物件所佔用的記憶體空間不大），開發人員可以隨時掃描ReferenceQueue，並對其中的Reference物件進行清除。
注意，一個物件同一時間並不一定只被另一個物件引用，而是可能被若干個物件同時引用。只要對這個物件的引用中有一個沒有使用“軟引用”特性，那麼垃圾回收策略對它的回收就不會採用“軟引用”的回收策略進行。如下圖所示：
上圖中，有兩個物件元素同時對B物件進行了引用（注意是同一個B物件，而不是對B類分別new了兩次）。其中A物件對B物件的依賴通過“軟引用”（SoftReference）間接完成，D物件對B物件的引用卻是通過傳統的“硬引用”完成的。當垃圾回收策略開始工作時它會發現這樣的情況，並且即使在記憶體空間不夠的情況下，也不會對B物件進行回收，直到針對B物件的所有引用可達性消失。
在JAVA中還有弱引用、虛引用兩個概念（Camel中的LRUWeakCache就是基於弱引用實現的）。但是由於他們至少和我們重點說明的DefaultCamelContext沒有太多關係，所以這裡筆者就不再發散性的講下去了。對這塊還不太瞭解的讀者可以自行參考JDK官方文件。
4-2-4、LRU演算法簡介
LRU的全稱是Least Recently Used（最近最少使用），它是一種選擇演算法，有的文章中也把LRU演算法稱為“快取淘汰演算法”。在計算機技術實踐中它被廣泛用於快取功能的開發，例如處理記憶體分頁與虛擬記憶體的置換問題，或者又像Camel那樣用於計算選擇Endpoint物件將從快取結構中被移除。下圖的結構說明了LRU演算法的大致工作過程：
上圖中，我們可以看到幾個關鍵點：
整個佇列有一個閥值用於限制能夠存放於佇列容器中的最大元素個數，這個閥值我們暫且稱為maxCacheSize。
當佇列中的元素還沒有達到這個maxCacheSize時，進入佇列的元素將被放置在佇列的最前面，佇列會保持這種處理策略直到佇列中的元素達到maxCacheSize為止。
當佇列中的某個元素被選擇時（一般來說，佇列允許開發人員在選擇元素時傳入一個Key，佇列會依據這個Key進行元素選擇），被命中的元素又會重新排列到佇列的最前面。這樣一來，佇列最尾部的元素就是近期使用最少的一個元素。
一旦當佇列中的元素達到maxCacheSize後（不可能超過），新進入佇列中的元素將會把佇列最尾部的元素擠出佇列，而它自己會排列到佇列的最頂部。
4-2-5、Camel中的LRUSoftCache
那麼我們介紹的SoftReference、LRU和我們本節正在講述的DefaultCamelContext有什麼聯絡呢？在DefaultCamelContext中，用來進行Endpoint註冊儲存管理的類稱為EndpointRegistry，它就是依據LRU演算法原則決定哪些Endpoint定義應該存放在快取中。具體來說，EndpointRegistry中使用“軟引用”方式，通過ConcurrentLinkedHashMap提供的既有LRU技術支援實現了存在於記憶體中的高效快取。它在DefaultCamelContext的變數定義如下：
......
private Map<EndpointKey, Endpoint> endpoints = new EndpointRegistry(this, endpoints);
......EndpointRegistry和它繼承的父類LRUSoftCache，以及它更高層的父類LRUCache的主要結構如下所示：
- LRUCache的主要結構
/**
* A Least Recently Used Cache.
* If this cache stores org.apache.camel.Service then this implementation will on eviction
* invoke the {org.apache.camel.Service#stop()} method, to auto-stop the service.
*/
public class LRUCache<K, V> implements Map<K, V>, EvictionListener<K, V>, Serializable {
// 這個值記錄LRU佇列的最大值
private int maxCacheSize = 10000;
// 一個布林型，表示如果LRU佇列中的元素被消除時，是否試著執行Service的stop方法
// 因為儲存在這個LRU中的元素一般來說是實現了Service介面的元素
private boolean stopOnEviction;
// 這個計數器用於統計LRU中元素的命中次數
private final AtomicLong hits = new AtomicLong();
// 這個計數器用於統計LRU中元素的未命中次數
private final AtomicLong misses = new AtomicLong();
// 這個計數器用於統計LRU中元素的移除數量
private final AtomicLong evicted = new AtomicLong();
// 由Google實現的一個數據結構，後文詳細介紹
private ConcurrentLinkedHashMap<K, V> map;
......
// 這個建構函式有三個引數，分別是：
// initialCapacity：LRU佇列的初始化大小
// maximumCacheSize：LRU佇列的最大元素大小
// stopOnEviction：這個布林值表示是否試圖對可能的Service元素進行stop操作
public LRUCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) {
// 建構函式主要的作用就是初始化ConcurrentLinkedHashMap物件
map = new ConcurrentLinkedHashMap.Builder<K, V>()
.initialCapacity(initialCapacity)
.maximumWeightedCapacity(maximumCacheSize)
.listener(this).build();
this.maxCacheSize = maximumCacheSize;
this.stopOnEviction = stopOnEviction;
}
......
/**
* 該方法在元素被從LRU佇列中登出時被觸發。
* 其中呼叫的stopService方法，將會試圖停止service的執行，如果value實現了Service介面的話
* */
@Override
public void onEviction(K key, V value) {
evicted.incrementAndGet();
LOG.trace("onEviction {} -> {}", key, value);
// 如果條件則開始stop動作
if (stopOnEviction) {
try {
// stop service as its evicted from cache
ServiceHelper.stopService(value);
} catch (Exception e) {
LOG.warn("Error stopping service: " + value + ". This exception will be ignored.", e);
}
}
}
......
}ConcurrentLinkedHashMap是Google提供的一個數據結構，其使用特性和java.util.LinkedHashMap一致，但是它是執行緒安全的。更重要的是ConcurrentLinkedHashMap的工作方式就是一個已經實現好的LRU的演算法。
- LRUSoftCache的主要結構
/**
* A Least Recently Used Cache which uses SoftReference.
*
* This implementation uses java.lang.ref.SoftReference for stored values in the cache
* to support the JVM when it wants to reclaim objects when it's running out of memory.
* Therefore this implementation does not support all the java.util.Map methods.
* */
public class LRUSoftCache<K, V> extends LRUCache<K, V> {
......
/**
* 建構函式
* */
public LRUSoftCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) {
// 這是呼叫父級LRUCache的建構函式
super(initialCapacity, maximumCacheSize, stopOnEviction);
}
......
/**
* SoftReference是LRUSoftCache最關鍵的地方，後文介紹
* */
@Override
@SuppressWarnings("unchecked")
public V put(K key, V value) {
SoftReference<V> put = new SoftReference<V>(value);
SoftReference<V> prev = (SoftReference<V>) super.put(key, (V) put);
return prev != null ? prev.get() : null;
}
......
/**
* 取出Key所對應的“軟引用”，並且從“軟引用”中檢視獲取value本身
* */
@Override
@SuppressWarnings("unchecked")
public V get(Object o) {
SoftReference<V> ref = (SoftReference<V>) super.get(o);
return ref != null ? ref.get() : null;
}
......
}SoftReference是LRUSoftCache最關鍵的地方，請注意以上程式碼片段中的put方法。該方法就是向ConcurrentLinkedHashMap送入一個新的K-V的元素，但是注意了，該方法並不是把Value直接送入ConcurrentLinkedHashMap，而是建立一個針對Value的“軟引用”SoftReference，並將其作為Value送入ConcurrentLinkedHashMap。通過get方法獲取Key對應的Value時，也是從ConcurrentLinkedHashMap中首先獲取“軟引用”物件。需要注意的是，這時的“軟引用”中是否還存在真實的值並不清楚，所以要進行一下判斷再進行返回。
- EndpointRegistry的主要結構
......
/**
* Endpoint registry which is a based on a {@link org.apache.camel.util.LRUSoftCache}.
* <p/>
* We use a soft reference cache to allow the JVM to re-claim memory if it runs low on memory.
*/
public class EndpointRegistry extends LRUSoftCache<EndpointKey, Endpoint> implements StaticService {
......
}
......實際上EndpointRegistry在程式碼結構中最主要的作用就是確定K-V的泛型結構，因為主要的LRU結構已經通過LRUCache實現了，另外基於“軟引用”的技術邏輯也都已經通過LRUSoftCache實現了。所以我們用一句話總結整個EndpointRegistry的實現：通過LRUCache保證已經註冊並且最近使用頻繁的Endpoint物件一定存在於快取中，通過LRUSoftCache保證所有已儲存在記憶體中Endpoint物件不會導致JVM記憶體溢位。
5、使用XML形式編排路由
除了上文中我們一直使用的DSL進行路由編排的操作方式以外，Apache Camel也支援使用XML檔案描述進行路由編排。通過XML檔案開發人員還可以將Camel和Spring結合起來使用——兩者本來就可以進行無縫整合。下面我們對這種方式的使用大致進行一下介紹。首先我們建立一個XML檔案，和Spring結合使用的：
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:camel="http://camel.apache.org/schema/spring"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring-2.14.1.xsd ">
<camel:camelContext xmlns="http://camel.apache.org/schema/spring">
<camel:endpoint id="jetty_from" uri="jetty:http://0.0.0.0:8282/directCamel"/>
<camel:endpoint id="log_to" uri="log:helloworld2?showExchangeId=true"/>
<camel:route>
<camel:from ref="jetty_from"/>
<camel:to ref="log_to"/>
</camel:route>
</camel:camelContext>
......
</beans>以上xml檔案中我們定義了一個Camel路由過程。請注意xml檔案中所使用的schema xsd路徑，不同的Apache Camel版本所使用的xsd路徑是不一樣的，這在Camel的官方文件中有詳細說明：http://camel.apache.org/xml-reference.html。在示例程式碼中筆者使用的Camel版本是V2.14.1。
XML檔案描述中，筆者定義了兩個endpoint：id為“jetty_from”的Endpoint將作為route的入口，接著傳來的Http協議資訊將到達id為“log_to”endpoint中。後者是一個Log4j的操作，最終Exchange中的In Message Body資訊將列印在控制檯上。接下來我們啟動測試程式：
......
/**
* 日誌
*/
private static final Log LOGGER = LogFactory.getLog(SpringXML.class);
public static void main(String[] args) throws Exception {
/*
* 這裡是測試程式碼
* 作為架構師，您應該知道在應用程式中如何進行Spring的載入、如果在Web程式中進行載入、如何在OSGI中介軟體中進行載入
*
* Camel會以SpringCamelContext類作為Camel上下文物件
* */
ApplicationContext ap = new ClassPathXmlApplicationContext("application-config.xml");
SpringXML.LOGGER.info("初始化....." + ap);
// 沒有具體的業務含義，只是保證主執行緒不退出
synchronized (SpringXML.class) {
SpringXML.class.wait();
}
}
......那麼在Camel中如何使用Spring中業已存在的Bean物件呢？我們再將本小計中以上示例進行深入，在Route中加入一個由Spring託管的處理器物件（Processor），並在Processor中引用Spring託管的另一個Bean物件：DoSomethingService。
- 這是一個由Spring容器管理的Bean。書寫方式就像您某個Spring工程中書寫一個Spring Bean一樣：
/**
* 這是一個服務層介面定義
* @author yinwenjie
*/
public interface DoSomethingService {
public void doSomething(String userid);
}
==================================以上是介面，下面是介面實現
/**
* 實現了定義的DoSomethingService介面，並且交由Spring Ioc容器託管
* @author yinwenjie
*/
@Component("DoSomethingServiceImpl")
public class DoSomethingServiceImpl implements DoSomethingService {
/**
* 日誌
*/
private static final Log LOGGER = LogFactory.getLog(DoSomethingServiceImpl.class);
/* (non-Javadoc)
* @see com.yinwenjie.test.cameltest.helloworld.spring.DoSomethingService#doSomething(java.lang.String)
*/
@Override
public void doSomething(String userid) {
DoSomethingServiceImpl.LOGGER.info("doSomething(String userid) ...");
}
}- 自定義的Processor處理器，交給Spring託管。可以看到和之前大家書寫的Processor沒有太大區別。無非是多出了一個Spring提供的“Component”註解標記：
/**
* 自定義的處理器，處理器本身交由Spring Ioc容器管理
* 並且其中注入了一個DoSomethingService介面的實現
* @author yinwenjie
*/
@Component("defineProcessor")
public class DefineProcessor implements Processor {
/**
* 日誌
*/
private static final Log LOGGER = LogFactory.getLog(DefineProcessor.class);
@Autowired
private DoSomethingService somethingService;
@Override
public void process(Exchange exchange) throws Exception {
// 呼叫somethingService，說明它正常工作
this.somethingService.doSomething("yinwenjie");
// 這裡在控制檯列印一段日誌，證明這個Processor正常工作了，就行
DefineProcessor.LOGGER.info("process(Exchange exchange) ... ");
}
}- 接下來我們就可以在原來的XML檔案中修改Route的編排，加入這個被Spring Ioc容器託管的Processor處理器。以上程式碼已經明示，這個自定義的Processor處理器在Spring Ioc容器中的id為“defineProcessor”：
......
<camel:camelContext xmlns="http://camel.apache.org/schema/spring">
<camel:endpoint id="jetty_from" uri="jetty:http://0.0.0.0:8282/directCamel"/>
<camel:endpoint id="log_to" uri="log:helloworld2?showExchangeId=true"/>
<camel:route>
<camel:from ref="jetty_from"/>
<camel:to ref="log_to"/>
<!-- 這是新加的processor處理器 -->
<camel:process ref="defineProcessor"></camel:process>
</camel:route>
</camel:camelContext>
......- 以下為控制檯顯示的執行效果：
[2016-07-11 19:37:36] INFO qtp405711462-19 Exchange[Id: ID-yinwenjie-240-55321-1468237049924-0-1, ExchangePattern: InOut, BodyType: org.apache.camel.converter.stream.InputStreamCache, Body: [Body is instance of org.apache.camel.StreamCache]] (MarkerIgnoringBase.java:96)
[2016-07-11 19:37:36] INFO qtp405711462-19 doSomething(String userid) ... (DoSomethingServiceImpl.java:24)
[2016-07-11 19:37:36] INFO qtp405711462-19 process(Exchange exchange) ... (DefineProcessor.java:31)注意控制檯列印的第一句日誌還是由原來的Log4j-endpoint列印的，接著路由會執行defineProcessor處理器中的somethingService.doSomething()方法，打印出第二句日誌。最後由defineProcessor中的Log4j打印出最後一句——整個由Camel和Spring整合的Route工作是正常的。