MassTransit探究初步繁体

MASSTRANSIT探究初步1背景作为SOA基礎設施，企業服務總線ESB是一個具有高分布性、事件驅動服務的SOA架構,是當前企業集成的主流框架。2簡介網站HTTP/GORRITWCOM/MASSTRANSITPROJECTCOM/MASSTRANSITMTISAFRAMEWORKFORCREATINGDISTRIBUTEDAPPLICATIONSONTHENETPLATFORMMTPROVIDESTHEABILITYTOSUBSCRIBETOMESSAGESBYTYPEANDTHENCONNECTDIFFERENTPROCESSINGNODESTHOUGHMESSAGESUBSCRIPTIONSBUILDINGACOHESIVEMESHOFSERVICES主要特性LBUSARCHITECTURELSAGASLEXCEPTIONMANAGEMENTLTRANSACTIONSLSERIALIZATIONLHEADERSLCONSUMERLIFECYCLELBUILTONTOPOFRABBITMQLIOCSUPPORT授權基於APACHE20，可以使用在任何環境中。3架構MASSTRANSIT在消息隊列（MQ）之上構建了消息總線機制，封裝了對消息隊列的操作，以及其它的組件，比如序列化、日志、SAGA、持久化等。下圖为傳入消息處理管道模型4整體分析MASSTRANSIT的目標是作为消息機制的抽象框架，因此，它本身並不具體實現MQ，而是通過集成其它MQ產品來作为其通信層。目前官方已集成的MQ產品有MSMQ、RABBITMQ。其它非官方補充了ACTIVEMQ。MASSTRANSIT在MQ之上添加了SAGAS、多線程、異常處理、事務、序列化、消息頭（HEADER）、消息使用者生命周期管理、路由、RX（REACTIVEEXTENSION反應式擴展）集成、NHIBERNATE集成、調試、跟蹤、日志輸出、加密、定時服務等。5特性分析51聲明式配置MASSTRANSIT本身使用了許多優秀的設計，比如對MSMQ、RABBITMQ的使用，通過在BUS構造配置中調用USEMSMQ或USERABBITMQ來聲明式的决定。52通過擴展方法隱藏具體實現通過分析源碼，可以發現USEXXX函數是通過定義在MASSTRANSITTRANSPORTSMSMQDLL和MASSTRANSITTRANSPORTSRABBITMQDLL中的擴展方法實現的，這样當用戶引用相應的DLL時，方法調用才會開放给用戶。即用戶選擇具體使用哪種TRANSPORT時，是通過引用相應的DLL來决定的。DLL引用入項目後，具體MQ產品特定的配置，函數等方法也附加到了基類對象中。這样可以避免在基類中定義大量子類特定接口MASSTRANSIT所有擴展組件均是采用這種方法，比如日志LOGGING組件。6入門首先需要定義消息PUBLICCLASSYOURMESSAGEPUBLICSTRINGTEXTGETSET消息總線的創建可以通過BUS類靜態方法INITIALIZE進行，傳入相應的配置方法，如下所示BUSINITIALIZESBCSBCUSEMSMQSBCVERIFYMSMQCONFIGURATIONSBCUSEMULTICASTSUBSCRIPTIONCLIENTSBCRECEIVEFROM“MSMQ/LOCALHOST/TEST“SBCSUBSCRIBESUBSSUBSHANDLERMSGCONSOLEWRITELINE“FROMBUS1“MSGTEXT初始化好BUS之後，就可以調用PUBLISH方法發布消息了BUSINSTANCEPUBLISHNEWYOURMESSAGETEXT“HI“在初始化中，通過SBCSUBSCRIBE函數注冊了一個HANDLER，其作用是在控制台中輸出消息文本。創建一個控制台應用程序，在MAIN函數中完成初始化和發布，可以在隨後的輸出中看到消息文本。7高級功能使用71SUBSCRIPTIONSERVICE参考HTTP/GORRITWCOM/DOCSMASSTRANSITPROJECTCOM/EN/LATEST/OVERVIEW/SUBSCRIPTIONSHTML如果使用的消息隊列不提供訂閱共享功能，比如MSMQ，可以使用MT的SUBSCRIPTIONSERVICE來實現此功能。在這種情況下，訂閱信息的協調功能由一個中心管理器來完成。這個中心管理器就是運行在網路中的SUBSCRIPTIONSERVICE的一個實例。每個消息總線（BUS）實例通過SUBSCRIPTIONCLIENT與其進行通信並交換訂閱信息。示例創建SUBSCRIPTIONSERVICEVARSUBSCRIPTIONBUSSERVICEBUSFACTORYNEWSBCSBCUSESTOMPSBCSETCONCURRENTCONSUMERLIMIT1SBCRECEIVEFROM“STOMP/LOCALHOST/MT_SUBSCRIPTIONS“VARSUBSCRIPTIONSAGASNEWINMEMORYSAGAREPOSITORYVARSUBSCRIPTIONCLIENTSAGASNEWINMEMORYSAGAREPOSITORYVARSUBSCRIPTIONSERVICENEWSUBSCRIPTIONSERVICESUBSCRIPTIONBUS,SUBSCRIPTIONSAGAS,SUBSCRIPTIONCLIENTSAGAS創建TIMEOUTSERVICEVARTIMEOUTBUSSERVICEBUSFACTORYNEWSBCSBCUSESTOMPSBCUSECONTROLBUSSBCRECEIVEFROM“STOMP/LOCALHOST/MT_TIMEOUTS“SBCUSESUBSCRIPTIONSERVICE“STOMP/LOCALHOST/MT_SUBSCRIPTIONS“VARTIMEOUTSERVICENEWTIMEOUTSERVICETIMEOUTBUS,NEWINMEMORYSAGAREPOSITORYTIMEOUTSERVICESTART創建應用BUSVARBUSSERVICEBUSFACTORYNEWSBCSBCUSESTOMPSBCUSECONTROLBUSSBCRECEIVEFROM“STOMP/LOCALHOST/YOUR_AWESOME_APPLICATION“SBCUSESUBSCRIPTIONSERVICE“STOMP/LOCALHOST/MT_SUBSCRIPTIONS“說明當調用USESUBSCRIPTIONSERVICE時，隱式附加了一個SUBSCRIPTIONCLIENT到BUS中。首先，SUBSCRIPTIONCLIENT會發送一個ADDSUBSCRIPTIONCLIENT消息给SUBSCRIPTIONSERVICE隊列，然後開始監聽訂閱變更，隨後發送ADDSCRIPTION/REMOVESUBSCRIPTION消息。通過這種方式，所有的更新將傳播到應用中所有其它節點中。72REQUEST/RESPONSE参考HTTP/GORRITWCOM/DOCSMASSTRANSITPROJECTCOM/EN/LATEST/OVERVIEW/REQUESTHTML通常使用MESSAGEBUS是一個應用發送一個請求（REQUEST），另一個應用接收到這個請求後做出一些回應（RESPONSE）。比如工資進程請求稅務進程執行某種所得稅，在計算完成後，返回結果。示例消息定義請求和響應通過CORRELATIONID來進行關聯。PUBLICCLASSBASICREQUESTCORRELATEDBYPUBLICGUIDCORRELATIONIDGETSETPUBLICSTRINGTEXTGETSETPUBLICCLASSBASICRESPONSECORRELATEDBYPUBLICGUIDCORRELATIONIDGETSETPUBLICSTRINGTEXTGETSET響應者簡單的在請求信息上添加一個RESP前綴作为響應。注意BASICRESPONSECORRELATIONID並沒有設置，這個是由MT自動設置的。PUBLICCLASSPROGRAMPUBLICSTATICVOIDMAINBUSINITIALIZESBCSBCUSEMSMQSBCVERIFYMSMQCONFIGURATIONSBCUSEMULTICASTSUBSCRIPTIONCLIENTSBCRECEIVEFROM“MSMQ/LOCALHOST/MESSAGE_RESPONDER“SBCSUBSCRIBESUBSSUBSHANDLERCXT,MSGCXTRESPONDNEWBASICRESPONSETEXT“RESP“MSGTEXT請求者提交請求並處理與原始請求相關聯（通過CORRELATIONID）的響應，處理後取消對響應的訂閱並結束這次請求。PUBLICCLASSPROGRAMPUBLICSTATICVOIDMAINBUSINITIALIZESBCSBCUSEMSMQSBCVERIFYMSMQCONFIGURATIONSBCUSEMULTICASTSUBSCRIPTIONCLIENTSBCRECEIVEFROM“MSMQ/LOCALHOST/MESSAGE_REQUESTOR“BUSINSTANCEPUBLISHREQUESTNEWBASICREQUEST,XXHANDLEMESSAGECONSOLEWRITELINEMESSAGETEXTXSETTIMEOUT30SECONDS上述代碼會阻塞調用線程直到響應返回或超時。超時後會拋出REQUESTTIMEOUTEXCEPTION。如果響應處理發生異常，該異常會重新拋出给調用請求發出線程。如果想使用異步處理，可以調用BEGINPUBLISHREQUEST，在處理結束後，調用ENDREQUEST方法完成處理。73SAGAS参考HTTP/GORRITWCOM/DOCSMASSTRANSITPROJECTCOM/EN/LATEST/OVERVIEW/SAGAHTML一個SAGA是指由協調器（COORDINATOR）管理的一個長事務。SAGAS通常由一個事件（EVENT）启動，SAGAS對事件進行編排，並維護所有事務的狀態。731定義SAGAS在MT中有兩種方法定義SAGA一種是用接口和類直接定義初始化、協調交互的各種消息，或可以被SAGA示例觀察到的各種消息；另一種是通過在類裏面定義事件、狀態、行为來定義一個狀態機來實現。732通過狀態機實現SAGAS要通過狀態機實現SAGAS，必須從SAGASTATEMACHINE類繼承PUBLICCLASSAUCTIONSAGASAGASTATEMACHINE,ISAGASTATICCOMBINESAGADEFINE/THESTATEMACHINEBEHAVIORISDEFINEDHEREPUBLICGUIDCORRELATIONIDGETSETPUBLICISERVICEBUSBUSGETSET然後定義各種狀態PUBLICSTATICSTATEINITIALGETSETPUBLICSTATICSTATECOMPLETEDGETSETPUBLICSTATICSTATEOPENGETSETPUBLICSTATICSTATECLOSEDGETSET定義與狀態關聯的事件PUBLICSTATICEVENTCREATEGETSETPUBLICSTATICEVENTBIDGETSET定義消息PUBLICINTERFACECREATEAUCTIONCORRELATEDBYSTRINGTITLEGETSTRINGOWNEREMAILGETDECIMALOPENINGBIDGETPUBLICINTERFACEPLACEBIDGUIDBIDIDGETGUIDAUCTIONIDGETDECIMALMAXIMUMBIDGETSTRINGBIDDEREMAILGET定義狀態轉換STATICAUCTIONSAGADEFINEINITIALLYWHENCREATEDURINGOPEN,WHENBID定義事件發生時的行为STATICAUCTIONSAGADEFINEINITIALLYWHENCREATETHENSAGA,MESSAGESAGAOPENINGBIDMESSAGEOPENINGBIDSAGAOWNEREMAILMESSAGEOWNEREMAILSAGATITLEMESSAGETITLETRANSITIONTOOPEN/PUBLICDECIMALOPENINGBIDGETSETPUBLICSTRINGOWNEREMAILGETSETPUBLICSTRINGTITLEGETSET完整的定義STATICSUPERVISORSAGADEFINEINITIALLYWHENCREATETHENSAGA,MESSAGESAGAPOSTALCODEMESSAGEPOSTALCODEPUBLISHSAGA,MESSAGENEWREQUESTPOSTALCODEDETAILSSAGAPOSTALCODEPUBLISHSAGA,MESSAGENEWREQUESTGEOLOCATIONSAGAPOSTALCODETRANSITIONTOWAITINGDURINGWAITING,WHENPOSTALCODEDETAILSRECEIVEDTHENSAGA,MESSAGESAGACITYMESSAGECITYSAGASTATEMESSAGESTATE,WHENGEOLOCATI