基於Http協議訂閱發佈系統設計

 

基於Http協議訂閱發佈系統設計

--物聯網系統架構設計

 

1,訂閱發佈(subscriber-publisher)

     訂閱發佈模式最典型的應用場景就是消息系統的設計。在消息系統的架構中，消息的發送者稱做(publisher)，消息的接收者稱做(subscriber)，參見wikipedia: Publish–subscribe pattern。整個消息系統的架構能夠用以下圖1來描述:

 

圖1

  由圖1可知消息系統主要包括3個組件: 發佈者，訂閱者和消息代理(Broker)，而整個消息系統的核心便是Broker，而目前就業務能力而言Broker的實現難點主要在於它的吞吐量。拿手機消息推送舉例，在當前的移動互聯時代，就咱們很常見的大多數app用戶數基本都是百萬級別以上（流行app基本是千萬級別），這意味着Broker至少要能支持百萬臺設備的訂閱，使用單臺服務器作Broker顯然不能解決問題。而在物聯網時代，訂閱者將再也不只有手機，訂閱者能夠是任何電子設備，這種場景的級別將是手機數量的百倍。

 

2,Mqtt協議的發佈訂閱系統實現方案

2.1，Mqtt協議

     根據官方的定義，mqtt協議便是 machine-to-machine (M2M)的鏈接協議，該協議就是爲發佈訂閱模式設計的很是輕量的消息傳輸協議。具體參見: http://mqtt.org/

從mqtt協議定義可知，該mqtt就是爲發佈訂閱系統而設計，而且很是輕量。

2.2，實現方案

     實現一套完整的發佈訂閱系統，主要就是兩個組件(client和broker)一個協議規範(mqtt)。

     目前流行的開源mqtt client實現是paho( http://eclipse.org/paho); 流行的開源mqtt broker實現包括 apache apollo 和  Eclipse Mosquitto( http://eclipse.org/mosquitto ), mosquitto的優勢是很是輕量，使用一臺樹莓派(或路由器)這樣的小型設備足夠服務一個家庭的設備鏈接。

2.3， 架構設計

     發佈訂閱的服務系統架構很是簡單，基本都遵守圖1的基本架構模式。對於一個家庭的物聯網應用，若是設備僅想要在局域網內訪問，則broker只須要安裝在(基於NanoPi或RasPi開發的)小型的設備中或者直接集成到路由器中。固然對於真正的物聯網應用，咱們仍是但願設備能夠經過互聯網就能夠管理和控制，因此不少broker實際應當在互聯網服務器中。

2.4， Mqtt協議的訂閱發佈系統交互原理

     首先引用一下開源項目paho提供的python版客戶端執行訂閱和發佈動做的demo，代碼很是簡短

 1 #susbscriber
 2 import paho.mqtt.client as mqtt
 3 
 4 # The callback for when the client receives a CONNACK response from the server.
 5 def on_connect(client, userdata, rc):
 6     client.subscribe("$SYS/#")
 7 
 8 # The callback for when a PUBLISH message is received from the server.
 9 def on_message(client, userdata, msg):
10     print(msg.topic+" "+str(msg.payload))
11 
12 client = mqtt.Client()
13 client.on_connect = on_connect
14 client.on_message = on_message
15 client.connect("iot.eclipse.org", 1883, 60)
16  
17 # Blocking call that processes network traffic
18 client.loop_forever()

View Code

 

     Subscriber: 從訂閱者客戶端代碼可知，訂閱者只需作2個動做（鏈接broker和創建循環等待的長鏈接）和提供2個接口函數（訂閱請求函數和處理broker響應結果的函數）。基本要素無非請求鏈接、訂閱指定topic消息、和處理響應結果，但loop_forever()是一個無限循環，這意味着客戶端和borker之間保持着一個socket長鏈接，因此從這裏能夠認識到broker的瓶頸之一即是能處理多少個這樣的長鏈接。

1 #publisher
2 import paho.mqtt.client as mqtt
3  
4 client = mqtt.Client()
5 client.connect("iot.eclipse.org")
6 client.loop_start()
7 res = mqttc.publish("$SYS/#", "HELLO")
8 client. loop_stop(force=False)

     Publisher: 從發佈者客戶端代碼可知，發佈者操做比訂閱者更加簡單，基本要素無非是創建鏈接、向broker發佈指定topic消息，忽略結果響應處理過程。

     subscriber和publisher的交互邏輯本質是基於tcp協議的socket實現，對於server端的socket打開mqtt協議端口，並開啓一個異步線程來持續監聽端口，等待client端(subscriber和publisher )的socket發出mqtt請求，client端的subscriber的mqtt請求有些不同，那就是subscriber的socket實際和server一直保持長鏈接，隨時等待server那邊推送過來的消息，直到鏈接關閉 。因此拋開細節處理問題，徹底可使用netty框架，基於mqtt協議很快的開發出一套server和client端的應用。

 

3,http協議broker設計實現

 

圖2 訂閱發佈系統Broker設計

  http協議和mqtt協議比較：

          優勢：http在互聯網時代獲得最普遍的應用， 充分檢驗了它的有效性和穩定性，充分的社區支持和成熟的開源資源可用

          缺陷：相對mqtt協議過重，對網絡要求更高，直接基於http1.x沒法實現發佈訂閱(http1.x是單工協議，須要依賴websocket、servlet3.0等技術實現雙工，也可使用http2.0，但目前支持較少)

 

 

   本文是使用servlet3.0的技術實現基於http協議的發佈/訂閱系統broker, 圖2所示即爲物聯網broker系統設計架構。後臺broker分紅兩大模塊：發佈中心(用戶和設備)和訂閱中心(用戶和設備)，以及事件總線。這樣的設計或許會有疑惑，爲何不直接抽象成事件的發佈和訂閱中心，如此不久和mqtt broker一致了麼？ 的確，既然是使用http協議實現，那爲何要徹底仿照mqtt協議的模式呢，並且咱們要設計的實際是一個「物聯網的業務系統「而不是一個「中間件「，因此若是你換了一個業務場景，你又得從新設計系統，而恰巧基於http協議servlet應用正是爲業務系統提供了豐富的開源資源。

 

  下面詳細解釋用戶發佈中心和訂閱中心的設計，由於在物聯網的應用場景中，主要業務交互邏輯是圍繞用戶和設備之間作publish和subscribe.

  用戶發佈中心(publisher):

     在物聯網場景中用戶充當了核心業務的publisher，對於broker的發佈中心，接收到全部的前端用戶請求過來的數據都將被封裝成event在broker的內部系統中由發佈中心廣播到訂閱中心。以摩拜單車爲例，app是publisher的終端，摩拜單車的核心業務邏輯就是開鎖指令和一系列的交易邏輯。就開鎖動做而言，發佈中心收到開鎖event，在publish這個event以前，針對這個event不一樣業務場景或許有不一樣的業務需求，典型場景有：該事件是否須要羣發、該事件是否須要定時功能，該事件是否須要可靠發佈。特別的，對於事件的可靠發佈，在交易類系統中屬於必備要求。拿摩拜單車來講，開鎖指令發出後就會開始計時計費和扣錢，這時候就須要依賴broker在應用層面對數據作事務保證，而不能依賴基礎系統服務的穩定。

  訂閱中心(subscriber):

      對於訂閱中心(不管是用戶或設備)的設計，徹底遵守table或key-value的數據結構來設計，也便是對於每個請求，broker都將爲其關聯一個handler以及和其對應id標識。當事件被髮布到訂閱中心，訂閱中心的processor便會用事件ID(或惟一標識的設備ID)去查詢對應的handler，並做結果響應。因爲是基於http協議，因此在具體實現時須要依賴servlet3.0或websoket技術。

 

4,領域建模

      4.1 發佈中心領域建模

     發佈中的核心功能是發佈事件，所以Event是發佈中心的核心領域對象。在圖2中已經闡明，事件發佈所須要實現的基本功能要素，Event設計也就主要是達到第3部分所描述功能。

圖3 發佈中心領域模型抽象

     

  在圖3中可知，AbstractEvent便是Event的頂層的Entity抽象設計。由於發佈中心可能會發布多種不一樣類型的Event，因此AbstractEvent必須有EventType屬性來表述事件的類型。不管是那種類型的Event實際都是一個Entity，既然是Entity就意味着有本身的ID，EventId做爲event的惟一標識符，須要有一個明確的說明的是EventId表示意義實際至關於topic，這就是說不是每發佈一個Event就會生成一個新的EventId。例如在摩拜單車的應用中，就開鎖這一類事件，對於每一輛單車，都有對應一個惟一的EventId。對於以前第3部分提到的關於事件須要實現週期、延時以及可靠發佈，AbstractEvent定義了cronExpression和deliveryStatus屬性，其中cron表達式能夠很是簡潔的描述和實現週期和延時的事件設定， 而deliveryStatus則須要使用狀態來保證分佈式網絡環境下事件動做的事務。此外，定義GroupEvent是爲了解決第3部分中提到的發佈一個事件，響應多臺設備。

      4.2 訂閱中心領域建模

  訂閱中心領域核心抽象是Handler，每個handler對應爲一個訂閱http request。每個訂閱請求handler都持有其但願響應的EventId，攜帶的業務數據以及結果響應的回調方法。訂閱者指望的是當EventId標識的event發生時，能夠馬上收到對應的事件響應，也便是說訂閱http request是做爲一個保持長時間等待的網絡鏈接。所以全部的handler應當有一個holder將其緩存起來管理，這就是單例模式的HandlerHolder存在的意義。對於HandlerHolder在對handler緩存策略能夠有兩種選擇：1， 以table形式緩存；2，以map形式緩存。相較兩種緩存策略各有優缺點，table形式節約存儲但查找代價高(可有序存放提升速度，實際在使用Java HashMap或ConcurrentHashMap實現的保持10^5個鏈接時，並不會多消耗太多內存，相對於List也僅僅是多幾十M的內存而已，由於map中的空桶實際存儲量極小)，map形式查找快但耗存儲，但不管那種形式緩存均可以經過分級緩存來提升緩存能力(例如一級緩存簡要數據在系統內存，二級緩存主要數據在redis等緩存系統)。

  在圖4中的狀態圖描述了(用戶和設備)訂閱中心以event驅動的handler轉移流程。初始時刻，設備發起訂閱CommandEvent請求，等待發布中心收到用戶發過來的CommandEvent請求，此時發佈中心會去判斷該事件是否須要記錄事件交付狀態，如須要獲得設備響應的OKEvent，則會去訂閱中心生成對應handler。此時，響應給設備的handler將攜帶deliveryStatus=Waiting 標識，等待設備返回確認結果。隨後，設備返回的確認響應便可經過發佈中心發佈OKEvent響應用戶處理結果。（實際處理流程應當更復雜，由於沒有考慮異常狀況，如設備沒有收到響應結果、備響應結果丟失等，這些都須要作一些補償策略）

 

圖4 訂閱中心領域建模抽象