1、概述

在我開始構思這幾篇關於“自己動手設計ESB中介軟體”的文章時，曾有好幾次動過放棄的念頭。原因倒不是因為對冗長的文章產生了惰性，而是ESB中所涉及到的技術知識和需要突破的設計難點實在是比較多，再冗長的幾篇博文甚至無法對它們全部進行概述，另外如果在思路上稍微有一點差池就會誤導讀者。一個可以穩定使用的ESB中介軟體凝聚了一個團隊很多參與者的心血，一個人肯定是無法完成這些工作的。但是筆者思索再三，還是下決心將這這即便文章完成，因為這是對本專題從第19篇文章到第39篇文章中所介紹的知識點的最好的總結。我們自己動手設計ESB中介軟體，不是為了讓它商用，也不是為了讓它可以比擬市面上某款ESB中介軟體，甚至不是為了把ESB中的技術難點的解決全部方案化。我們的目的是檢驗整個專題中所介紹的知識點是否能在讀者自己消化後進行綜合應用，是否能做到技術知識的活學活用、按需選型

2、ESB的頂層設計

（頂層設計圖）

上圖是我們要進行實現的ESB中介軟體的頂層設計。從上圖中可以看到，整個ESB中介軟體分為以下幾個模組：Client客戶端、流程編排/註冊工具、主控服務模組、服務狀態協調組（模組）、服務執行組（模組）。首先我們大致描述一下這些模組的工作內容：

• Client客戶端是需要接入ESB中介軟體的各個業務服務系統。例如物流系統、聯賬系統、CRM系統等等。在這些客戶端系統接入ESB中介軟體時，將整合ESB中介軟體提供給他們的各種開發語言版本的ESB-Client元件。如果使用的是C#語言則ESB-Client元件可能以DLL檔案的方式提供；如果使用的是Java語言則ESB-Client元件可能以Jar檔案的方式提供；如果使用的是NODEJS則可能是一個（或多個）JS檔案……

• 這些客戶端系統的開發人員將可以使用ESB中介軟體提供的一個獨立的流程編排/註冊工具，後者在很多ESB中介軟體系統中一般被命名為“…… Studio”，並且這些流程編排/註冊工具一般以各種IDE外掛的形式提供出來，例如製作成Eclipse-Plugin提供給開發人員。這些工具的主要作用就是讓客戶端系統的開發人員（開發團隊）具備向ESB主控服務進行原子服務註冊的能力，另外還可以讓開發人員查詢到目前服務端所有可用的其它原子服務（來自於其它業務系統的），以便在流程編排/註冊工具上完成新的服務流程編排和已有服務流程新版本的釋出。這就是上圖中標註為“1”的步驟。

• 另外ESB中介軟體為了保證流程編排所使用的原子服務不會因為提供這個原子服務的業務系統的變化而產生影響，一般來說在進行業務系統註冊原子服務時都會指定這個原子服務的版本和呼叫許可權。呼叫許可權一般又分為黑名單許可權和白名單許可權。以白名單許可權來說，只有白名單中所列列舉的業務系統有許可權呼叫這個原子服務。即使這個原子服務參與了某個ESB中的流程編排，如果請求這個編排好的流程的業務系統不在這個白名單中，呼叫也會失敗。

• 主控服務為流程編排/釋出工具提供新的原子服務註冊請求、新的流程釋出請求、已有流程的新版本釋出請求。最新的原子服務、流程編排等資料將會被主控服務儲存在持久化容器中（例如關係型資料庫），並且向“服務狀態協調模組”傳送最新的資料變化。注意，主控服務並不負責執行編排好的流程，只是用於記錄資料編排的變化和向“資料協調模組”傳送這些資料變化，這就是上圖中所標示的步驟2。主控服務還有另外兩個作用：負責許可權管理和服務執行模組的狀態監控。

• 由於負責最終對流程編排進行執行的“服務執行模組”存在很多節點（下文稱為ESB-Broker Server節點），且這些ESB-Broker Server節點的數量在服務過程中會不斷變化（新增或減少），所以“主控服務”並不知道有哪些Boker在執行。為了通知這些在執行狀態的Broker有新的服務編排被髮布（或者其它事件資訊），這些處於執行狀態的ESB-Broker Server節點都會連線到“服務狀態協調模組”，並且由後者完成資料變化的事件通知。這就是“服務狀態協調模組”的主要功能，也是上圖中所示的步驟3。在我們自己設計的ESB中介軟體中，“服務狀態協調模組”由一組zookeeper服務構成（在我另外幾篇博文中專門介紹zookeeper，這個專題就不對zookeeper的基本操作進行講解了），如果您在實際的工作中有其它功能/技術需求，也可以自己設計“服務狀態協調模組”。

• 在業務系統整合過程中，ESB中介軟體所扮演的角色就是在各個業務系統間進行原子服務呼叫、轉換資料、再進行原則服務呼叫、再轉換…….最後向執行服務編排的請求者返回結果。所以ESB中介軟體服務往往有較高的效能要求。如果執行ESB服務編排的節點只有一個，往往就達不到ESB中介軟體的設計要求甚至會使ESB中介軟體服務成為整個軟體架構的效能瓶頸點。所以在我們設計的ESB中介軟體中，真正執行ESB服務的節點會有多個這種ESB-Broker Server節點。

• 在ESB執行服務的過程中使用多個Broker Server有很多好處，首先來說它們可以保證在整個系統出現請求洪峰的情況下，能夠把這些請求壓力平均分配到這些Broker Server節點上，最終使ESB服務不會成為整個頂層設計的瓶頸。請求壓力的分配工作會由zookeeper叢集完成。另外，多個Broker Server可以保證某一個（或者幾個）Broker Server節點在出現異常並退出服務後，整個ESB中介軟體的服務不會停止——這是一個現成的容錯方案。開發人員可以通過退避演算法來決定ESB-Client下一次試圖訪問出現錯誤的Broker Server節點的時間，也可以立即為ESB-client重新分配一個健康的Broker Server節點。最後，這個解決方案可以在ESB服務執行的過程中保證實現Broker Server的動態橫向擴充套件：當ESB主控服務模組發現整個Broker Server服務組的效能達到（或快要達到）峰值時，運維人員可以馬上開啟新的Broker Server節點，zookeeper叢集會負責將定製的編排、定製的Processor處理器等資料資訊動態載入到新的Broker Server節點中，並讓後者立即加入整個服務組開始工作。

• 在ESB-Client（某個業務系統）請求執行某個服務編排時，首先會使用這個ESB-Client（某個業務系統）已經整合的zookeeper客戶端請求ESB的zookeeper叢集服務，從中取得當前正在執行的Broker Server節點資訊，並通過某種演算法決定自己訪問哪一個Broker Server節點（演算法很多：輪詢演算法、加權演算法、一致性Hash演算法等等），如“頂層設計圖”中步驟4、步驟5所示。為了保證上文中提到的新的Broker Server節點能夠加入服務組併為ClientESB-Client服務，步驟4和步驟5的過程可以週期性進行，並視情況重新為ESB-Client分配Broker Server節點。

• 當ESB-Client確定目標Broker Server節點後，將正式向這個Broker Server發起執行某個服務編排的請求。當同一個ESB-Client第二次請求執行服務編排時，就可以在一定時間週期內（有效時間內）不再走步驟4、5了，而可以直接發起請求到同一個目標Broker Server節點。直到這個Broker Server不再能夠響應這些請求為止（或者有其它依據確定這個Broker Server節點已經不能提供服務），ESB-Client會再執行步驟4、5，以便確定另一個新的、正常工作的Broker Server節點。在下文筆者也會重點介紹如何進行Broker Server節點的選擇。

3、主控服務的日誌收集

上一節已經說到，在我們設計的ESB中介軟體中包括兩個模組：主控服務模組和服務執行組（模組）。其中主控服務模組的其中一個作用，是對若干當前處於執行狀態的服務執行組節點（Broker Server）進行效能狀態監控。效能狀態監控的目的是確保運維人員實時瞭解這些Broker Server的執行狀態，並且能在整個服務執行組快要達到效能瓶頸時能夠啟動新的Broker Server分擔壓力或者在整個服務執行組沒有什麼請求負擔時，停止一些Broker Server。

那麼主控節點如何知道整個Broker Server組中多個服務節點的效能狀態呢？要知道，Broker Server節點是可以動態擴充套件的。上文也已經說到：主控節點並不知道當前有哪些Broker Server節點處於執行狀態。那麼基於Kafka訊息佇列的日誌收集就是一個解決方案，設計人員還可以使用Flume + Storm的解決方案進行日誌自動收集和即時分析。下面我們對這兩種日誌收集方案進行介紹。注意，關於Kafka、Flume在這個專題之前的文章中已經做了詳細介紹，所以本小節中涉及Kafka、Flume技術的部分就不再對設計方案的實施進行介紹了。

3-1、使用Kafka收集效能資料

Kafka Server的特點就是快，雖然在特定的情況下Kafka Server會出現訊息丟失或重複傳送的問題，但是這個問題針對日誌資料收集場景來說不算大問題。使用訊息佇列收集各Broker Server節點的效能日誌也是和ESB中各模組的依賴特性相適應的：由於在我們設計的ESB中介軟體中，主控服務模組並不知道有多少Broker Server節點處於執行狀態，也不知道這些Broker Server節點的IP位置。也就是說主控服務模組無法主動去這些Broker Server節點上收集效能資料。最好的辦法就是由這些活動的Broker Server節點主動傳送日誌資料。

3-1-1、設計思路

下圖是使用Kafka元件收集Broker Server節點上效能資料並進行效能資料處理、結果儲存的設計示例圖：

上圖中，每一個Broker Server節點上除了啟動一個Camel Context例項以外（後文進行詳細說明），還需要配置一個Kafka的Producer端用於傳送資料。Kafka-Producer端收集的效能資料可能包括：CPU使用情況、記憶體使用情況、本地I/O速度、作業系統執行緒情況、Camel Context中的路由例項狀態、Endpoint在Cache中的命中情況、客戶端對Broker Server中以編排路由的呼叫情況等等——業務資料和非業務資料都可以通過這種方式進行監控，並且以業務資料為主。

Kafka Servers中部署了三個Kafka Broker Server節點（建議的值），用於接收若干ESB Broker Server節點上各個Kafka-Producer傳送來的效能日誌資料。為了保證整個Kafka叢集的效能，每一個Kafka Broker Server都有至少兩個分割槽（partition，還是建議值）。這裡多說一句，為了節約服務資源您可以將Kafka Broker Server和Kafka-Consumer放在一臺服務節點上，甚至可以將它們和主控服務節點放在一起。

Kafka-Consumer負責進行效能日誌資料的處理。有的讀者可能就要問了，既然Consumer接收到的都是可以獨立儲存效能日誌資料，那麼只需要將這些日誌找到一個合適的儲存方案（例如Hbase）存放起來就可以了，還需要Consumer做什麼處理呢？這是因為開發團隊完成的Producer取樣頻率可能和運維團隊要求的監控採用頻率不一樣。

為了保證效能監控資料的精準性，開發團隊利用基於Kafka叢集提供的吞吐量優勢，可以在各個ESB Broker Server節點所整合的Kafka-Producer上設定一個較高的取樣率（當然還是要顧忌節點本身的資源消耗），例如每秒對固定的業務指標和非業務指標完成10次取樣。但是運維團隊通過主控服務監控各個ESB Broker Server節點是，往往不需要這麼高的取樣率（這裡可以提供一個設定選項供運維團隊隨時進行調整），大概也就是每秒更新1次的樣子就差不多了。

那麼Consumer如何處理每秒鐘多出來的9次取樣資料呢？可以明確想到的有兩種處理方式：一種處理方式是無論主控服務的監控臺上的效能指標以何種頻率進行顯示，Consumer都將收到的資料寫出儲存系統中；另一種處理方式是Consumer將收到的多餘資料丟棄，只按照運維團隊設定的取樣頻率將資料寫入持久化儲存系統。在第二中處理方式中有一個情況需要特別注意：如果將要被丟棄的效能資料達到了效能閥值（例如本次採集的記憶體使用率超多了2GB），則這條日誌資料還是需要進行保留。第一種處理基本上沒有什麼需要介紹的，優點和缺點也是很明確的：優點是可以在後期進行完整的效能歷史回溯，缺點就是會佔用較大的儲存空間——雖然目前可以使用的超大儲存方案有很多而且都很成熟穩定，但它們都需要比較強大的資金預算支援。

3-1-2、Consumer的實現

這裡筆者主要討論一下Consumer的第二種處理方式：丟棄多餘的資料。我們可以使用之前文章介紹過的ConcurrentLinkedHashMap作為Consumer中儲存效能訊息日誌的Cache，Cache的固定大小設定為200（或者其它一個較大的值）。這個Cache結構可以幫助我們完成很多工作：首先它可靠的效能能夠保證過個Consumer不會成為整個效能日誌收集方案的瓶頸——雖然ConcurrentLinkedHashMap的效能並不是最快的；其次這個Cache結構能夠幫助我們自動完成多餘效能日誌的清除工作，因為在第201條日誌記錄被推入Cache時，在LRU佇列尾部的最初一條記錄將自動被排除佇列，最終被垃圾回收策略回收掉；最後，Consumer按照運維團隊設定的取樣週期，對Cache中的效能日誌資料進行持久化儲存時，始終只需要取出當前在Cache將被剔除的那條記錄，這樣就省掉了編寫程式，在兩個週期的時間差之間判斷“要對哪條效能日誌資料”進行持久化儲存的定位工作。

順便說一句，如果您需要在工程中使用Google提供的ConcurrentLinkedHashMap資料結構工具，那麼您需要首先在pom檔案中新增相應的元件依賴資訊：

<dependency>
    <groupId>com.googlecode.concurrentlinkedhashmap</groupId>
    <artifactId>concurrentlinkedhashmap-lru</artifactId>
    <version>1.4.2</version>
</dependency>

• 1
• 2
• 3
• 4
• 5

• 1
• 2
• 3
• 4
• 5

以下是Consumer中用於處理LRU佇列新增、LRU週期性讀取、LRU刪除事件的程式碼片段：

......
/**
 * 這就是效能資料的LRU佇列
 */
private static final ConcurrentLinkedHashMap<Long, String> PERFORMANCE_CACHE = 
        new ConcurrentLinkedHashMap.Builder<Long, String>()
        .initialCapacity(200)
        .maximumWeightedCapacity(200)
        .listener(new EvictionListenerImpl())
        .build();
......

/**
 * 這個監聽器用於在資料被從LRU佇列剔除時<br>
 * 按照功能需求檢查這條記錄是否需要被持久化儲存起來。
 * @author yinwenjie
 */
public static class EvictionListenerImpl implements EvictionListener<Long, String> {

    // 上一次進行資料採集的時間，初始為-1
    private Long lastTime = -1l;

    // 這是由運維團隊設定的資料採集週期，1000表示1000毫秒
    // 正式系統中，這個值將有外部讀取
    private Long period = 1000l;

    @Override
    public void onEviction(Long key, String jsonValue) {
        /*
         * 以下條件任意成立時，就需要對這條資料進行採集了：
         * 1、lastTime為-1的情況（說明是程式第一次採集）
         * 
         * 2、當前事件 - lastTime >= period(採集週期)
         * 
         * 3、當監控資料大於設定的警告閥值，在這個示例程式碼中
         * 這個警告閥值為80，正式系統中，這個閥值應從外部讀取
         * 以下的threshold變數就代表這個值
         * */
        Long threshold = 80L;
        Long nowtime = new Date().getTime();
        // 獲取效能資料中的CPU使用率
        // 注意，正式系統 中最好不要傳遞json結構，文字結構的資料就好了
        JSONObject jsonData = JSONObject.fromObject(jsonValue);
        Long cpuRate = jsonData.getLong("cpu");
        boolean mustCollecting = false;
        if(this.lastTime == -1 || 
            nowtime - lastTime >= this.period ||
            cpuRate >= threshold) {
            mustCollecting = true;
            this.lastTime = nowtime;
        }
        // 如果不需要做資料的持久化儲存，就終止本次監聽的操作即可
        if(!mustCollecting) {
            return;
        }

        // ********************
        // 這裡可以做持久化資料儲存的操作了
        // ********************
        LRUConsumer.LOGGER.info(key + ":" + jsonValue + " 完成資料持久儲存操作=======");
    }
}

......

// 以下程式碼就是當Kafka-Consumer收到效能日誌資料的操作
// 將這個資料存放到PERFORMANCE_CACHE即可
Long key = new Date().getTime();
// 可使用時間的毫秒數作為key值（正式應用場景下，考慮多個consumer節點，Key的確定會有一個更規範的規則）
LRUConsumer.PERFORMANCE_CACHE.put(key, performanceData);
......

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71

以上程式碼中，我們使用之前已經介紹過的LRU資料結構在Consumer端儲存傳送過來的資料。如果讀者對LRU還不清楚可以檢視我另外的一篇文章中的介紹（《架構設計：系統間通訊（39）——Apache Camel快速入門（下2）》）。由Google提供的ConcurrentLinkedHashMap結構就可以向我們提供一個現成的LRU佇列，這樣一來當LRU佇列儲存滿後，最先被接收到的效能日誌資料就會從佇列尾部被刪除。最關鍵的處理工作都將在EvictionListener介面的實現類中完成，在實際應用中開發人員還可以在確定一條效能日誌需要被持久化儲存之後專門啟動一個執行緒進行操作，例如使用一個專門的執行緒池（ThreadPoolExecutor）。這樣一來LRU佇列就真正不受持久化儲存操作延遲時間的影響了。

3-2、使用Flume + Storm收集效能資料

以上使用Kafka收集Broker Server節點的效能資料的方案中，需要在每個編寫的Broker Server節點上增加額外的程式碼向Kafka Broker Server傳送資料。實際上這種功能需求情況使用Apache Flume收集資料會使技術方案更容易實現和維護，下面我們就大致介紹一下這個技術方案實現。由於在之前的文章中筆者已經較詳細的介紹瞭如何使用Apache Flume進行基本配置了，所以這裡我們重點討論兩個問題Apache Flume的資料來源和Storm Server在接收到Flume Server傳送來的資料後如何進行處理。

3-2-1、設計思路

上圖展示了整個功能需求的設計結構。安裝在ESB Broker Server節點的Flume程式負責收集這個節點上的各種功能性指標和非功能性指標，這樣避免了在ESC Broker Server服務上編寫額外的程式碼採集非功能性指標，也減少了編寫程式碼採集功能性指標的複雜度。然後將這些效能日誌資料按照負載均衡模式傳遞到若干中繼Flume Server節點上，後者專門用於承載/彙總多個ESB Broker Server節點傳來的效能日誌資料，並且最終將資料寫入Storm Server。在Flume Server和Storm Server之間我們還是需要使用Kafka Server作為快取，這是因為Apache Kafka通過Storm-Kafka元件和Storm Server實現無縫整合。

首先請注意安裝在ESB Broker Server節點的Flume程式，在3-1小節中採集節點功能性指標和非功能性指標都是依靠開發人員編寫程式完成，併發送給Kafka-Broker。但這樣做卻真的繞了很大一個彎路，因為Linux作業系統上已經提供了很多采集節點非功能性指標的方式（例如採集I/O資訊、記憶體使用資訊、記憶體分頁資訊、CPU使用資訊、網路流量資訊等），開發人員只需要一些指令碼就可以完成採集工作。例如，我們採集CPU資訊完全不需要我們在ESB-Broker Server中編寫程式（採集CPU資訊也不應該是ESB-Broker Server的一項工作任務），而採用如下的指令碼即可：

top -d 0.1 | grep Cpu >> cpu.rel

#寫法還有很多，還可以從/proc/stat檔案中獲取CPU狀態

• 1
• 2
• 3

• 1
• 2
• 3

以上指令碼可以按照100毫秒為週期，獲取CPU的資訊。並將這條資訊作為一條新的記錄儲存到cpu.rel檔案中。這樣Apache Flume就可以讀取cpu.rel檔案中的變化，作為效能日誌資料的來源：

#flume 配置檔案中的片段
......
agent.sources.s1.type = exec
agent.sources.s1.channels = c1
agent.sources.s1.command = tail -f -n 0 /root/cpu.rel
......

• 1
• 2
• 3
• 4
• 5
• 6

• 1
• 2
• 3
• 4
• 5
• 6

在ESB-Broker Server節點中，我們可以使用這樣的方式從不同檔案中讀取各種不同的日誌資訊，如下圖所示：

=================================
（接下文）