分享一個分佈式消息總線，基於.NET Socket Tcp的發佈-訂閱框架，附代碼下載

時間 2019-11-06

標籤分享一個分佈式消息總線基於 socket tcp 發佈訂閱框架代碼下載欄目系統架構简体版

原文原文鏈接

1、分佈式消息總線

在不少MIS項目之中都有這樣的需求，須要一個及時、高效的的通知機制，即好比當使用者A完成了任務X，就須要當即告知使用者B任務X已經完成，在一般的狀況下，開發人中都是在使用者B所使用的程序之中寫數據庫輪循代碼，這樣就會產品一個很嚴重的兩個問題，第一個問題是延遲，輪循機制要定時執行，必須會引發延遲，第二個問題是數據庫壓力過大，當進行高頻度的輪循會生產大量的數據庫查詢，而且若是有大量的使用者進行輪循，那數據庫的壓力就更大了。html

那麼在這個時間，就須要一套能支持發佈-訂閱模式的分佈式消息總線，那這個問題就能夠很好的解決了，好比採用一些成熟的消息總線進行實現，好比MSMQ或者採用好比開源的NServiceBus的發佈訂閱機制就能夠實現處理這種需求，其系統結構就會變成以下所示：git

本分佈式消息總線，目前普遍的被應用於分佈式緩存的更新通知，當在N百臺客戶短在使用緩存的過程之中，某個操做修改了緩存的數據，必須會致使其餘終端緩存的失效，那麼使用基於Socket的分佈式消息總線以後，咱們能夠作了修改了便可實時通知，作到緩存數據保持最新，再好比醫療應用之中的危急值管理，當發現檢驗、檢查危急值以後，須要及時通知病區啓動聲光報警系統等，提醒醫護工做人員及相關領導作出相應的措施，再好比應用於異構系統整合，當檢驗系統作出檢驗報告，經過消息總線進行發佈，HIS系統則即時會收到檢驗報告數據而實現系統的整合。github

2、基於Socket的實現

目前可以實現發佈訂閱模式的開源產品很是之多，爲何還要製造輪子呢，其主要緣由有如下幾點數據庫

1）像NServiceBus這種東西基於MSMQ，在大量的發佈者-訂閱者的狀況下性能不佳。緩存

2）此類東西太過於龐大、不易使用和配置。服務器

3）學習成本太高。併發

那爲何要使用Socket技術進行實現呢，其主要緣由是有如下幾點:app

1）使用高效的Socket通訊技術，高效、支持更多的客戶端。框架

2）使用簡單，不須要定義太多額外的東西，只須要定義主題和消息便可使用。socket

目前本發佈訂閱框架是基於AgileEAS.NET SOA中間件平臺Socket框架實現的，有關於些Socket框架的技術細節請參考AgileEAS.NET SOA 中間件平臺.Net Socket通訊框架-介紹、AgileEAS.NET SOA 中間件平臺.Net Socket通訊框架-簡單例子-實現簡單的服務端客戶端消息應答、AgileEAS.NET SOA 中間件平臺.Net Socket通訊框架-完整應用例子-在線聊天室系統-下載配置、AgileEAS.NET SOA 中間件平臺.Net Socket通訊框架-完整應用例子-在線聊天室系統-代碼解析文章進行了解和學習。

目前本發佈訂閱框架並直接集成於AgileEAS.NET SOA Socket通訊框架之中而且隨其一併發佈，下面簡單介紹一下其API：

在本框架之中定義了一個消息總線接口IMessageBus：

 
using System; 
   
using System.Collections.Generic; 
   
using System.Linq; 
   
using System.Text; 
   
using System.Collections; 
   
   
   
namespace EAS.Messages 
   
{ 
   
    /// <summary> 
   
    /// 消息總線接口定義。 
   
    /// </summary> 
   
    public interface IMessageBus : IDisposable 
   
    { 
   
        /// <summary> 
   
        /// 註冊發佈者。 
   
        /// </summary> 
   
        /// <param name="publisher">發佈者。</param> 
   
        void AddPublisher(string publisher); 
   
   
   
        /// <summary> 
   
        /// 註冊發佈者。 
   
        /// </summary> 
   
        /// <param name="publisher">發佈者。</param> 
   
        /// <param name="topic">主題。</param> 
   
        void AddPublisher(string publisher, string topic); 
   
   
   
        /// <summary> 
   
        /// 發佈一條消息到總線。 
   
        /// </summary> 
   
        /// <param name="topic">主題。</param> 
   
        /// <param name="message">發佈的消息。</param> 
   
        void Publish(string topic, object message); 
   
   
   
        /// <summary> 
   
        /// 訂閱消息。 
   
        /// </summary> 
   
        /// <param name="subscriber">訂閱者。</param> 
   
        /// <param name="topic">主題。</param> 
   
        /// <param name="notifyHandler">訂閱通知。</param> 
   
        void Subscribe(object subscriber, string topic, MessageNotifyHandler notifyHandler); 
   
   
   
        /// <summary> 
   
        /// 訂閱消息。 
   
        /// </summary> 
   
        /// <param name="subscriber">訂閱者。</param> 
   
        /// <param name="friendName">訂閱者名稱，用於處理離線訂閱。</param> 
   
        /// <param name="topic">主題。</param> 
   
        /// <param name="notifyHandler">訂閱通知。</param> 
   
        void Subscribe(object subscriber,string friendName ,string topic, MessageNotifyHandler notifyHandler); 
   
   
   
        /// <summary> 
   
        /// 訂閱消息。 
   
        /// </summary> 
   
        /// <param name="subscriber">訂閱者。</param> 
   
        /// <param name="friendName">訂閱者名稱，用於處理離線訂閱。</param> 
   
        /// <param name="topic">主題。</param> 
   
        /// <param name="notifyHandler">訂閱通知。</param> 
   
        /// <param name="changedHandler">發佈者狀態變化委託。</param> 
   
        void Subscribe(object subscriber, string friendName, string topic, MessageNotifyHandler notifyHandler,PublisherSstatusChangedHandler changedHandler); 
   
   
   
        /// <summary> 
   
        /// 退訂消息。 
   
        /// </summary> 
   
        /// <param name="subscriber">訂閱者。</param> 
   
        void Unsubscribe(object subscriber); 
   
   
   
        /// <summary> 
   
        /// 退訂消息。 
   
        /// </summary> 
   
        /// <param name="subscriber">訂閱者。</param> 
   
        /// <param name="topic">主題。</param> 
   
        void Unsubscribe(object subscriber, string topic); 
   
   
   
        /// <summary> 
   
        /// 退訂消息。 
   
        /// </summary> 
   
        /// <param name="subscriber">訂閱者。</param> 
   
        /// <param name="friendName">訂閱者名稱，用於處理離線訂閱。</param> 
   
        /// <param name="topic">主題。</param> 
   
        void Unsubscribe(object subscriber, string friendName, string topic); 
   
    } 
   
}

IMessageBus就基於Socket發佈訂閱消息總線的靈魂接口，也是基惟一的發佈者調用者功能入口，也就是說無論你是發佈者仍是訂閱者都須要調用這個接口，若是你是發佈者請調用IMessageBus接口的Publish方法向消息總線發佈消息，若是是你訂閱者請經過IMessageBus的訂閱方法進行訂閱，當你訂閱了某個主題以後，有發佈者發佈該主題的消息，你便可以收到消息並調用訂閱回調函數進行處理。

3、實現一個簡單的例子

如今咱們開始一個簡單的應用消息總線的例子，本例子代碼解決方案由下圖4個項目組成：

其中：Demo.Messages項目定義發佈者、訂閱者所使用的消息對象和消息主題。

Demo.Publisher項目爲發佈者代碼。

Demo.Subscriber項目爲訂閱者代碼。

Demo.Server項目爲服務端代碼。

在Demo.Messages項目之中，咱們定義了一個消息Message：

 
using System; 
   
using System.Collections.Generic; 
   
using System.Linq; 
   
using System.Text; 
   
using System.Xml.Serialization; 
   
   
   
namespace Demo.Messages 
   
{ 
   
    [Serializable] 
   
    public class Message 
   
    { 
   
        [XmlElement] 
   
        public Guid ID 
   
        { 
   
            get; 
   
            set; 
   
        } 
   
    } 
   
}

消息Message很簡單，只有一個屬性ID，同時還須要定義一個消息主題：

 
using System; 
   
using System.Collections.Generic; 
   
using System.Linq; 
   
using System.Text; 
   
   
   
namespace Demo.Messages 
   
{ 
   
    public class Topics 
   
    { 
   
        public static readonly string DEMO_TOPIC = "演示消息"; 
   
    } 
   
}

咱們定義了一個消息主題爲「演示消息」。

在Demo.Publisher項目之中，沒有太多額外的代碼，只有在Program.cs寫了如下簡單的調用代碼：

 
using System; 
   
using System.Collections.Generic; 
   
using System.Linq; 
   
using System.Text; 
   
using EAS.Messages; 
   
   
   
namespace Demo.Publisher 
   
{ 
   
    class Program 
   
    { 
   
        static void Main(string[] args) 
   
        { 
   
            var container = EAS.Objects.ContainerBuilder.BuilderDefault(); 
   
            var bus = container.GetComponentInstance("MessageBus") as IMessageBus; 
   
            System.Console.WriteLine("Publisher"); 
   
   
   
            while (System.Console.ReadLine()!="exit") 
   
            { 
   
                var m = new Messages.Message { ID = Guid.NewGuid() }; 
   
                bus.Publish(Demo.Messages.Topics.DEMO_TOPIC, m); 
   
                System.Console.WriteLine(string.Format("Publish:{0}", m.ID)); 
   
            } 
   
        } 
   
    } 
   
}

從IOC容器獲取一個消息總線IMessageBus對象，並調用Publish函數發佈消息」bus.Publish(Demo.Messages.Topics.DEMO_TOPIC, m);「。

固然了，使用了IOC容器，就離來開配置文件了，其App.config文件內容以下：

 
<?xml version="1.0" encoding="utf-8"?> 
   
<configuration> 
   
  <configSections> 
   
    <section name="eas" type="EAS.ConfigHandler,EAS.MicroKernel" /> 
   
  </configSections> 
   
  <eas> 
   
    <objects> 
   
      <!--消息總線--> 
   
      <object name="MessageBus" assembly="EAS.MicroKernel" type="EAS.Sockets.Bus.SocketBus" LifestyleType="Singleton"> 
   
        <property name="Url" type="string" value="socket.tcp://127.0.0.1:6606/"/> 
   
      </object> 
   
    </objects> 
   
  </eas> 
   
</configuration>

在Demo.Subscriber項目之中，使用與Demo.Publisher如出一轍的配置文件，其Program.cs代碼以下：

 
using System; 
   
using System.Collections.Generic; 
   
using System.Linq; 
   
using System.Text; 
   
using EAS.Messages; 
   
   
   
namespace Demo.Subscriber 
   
{ 
   
    class Program 
   
    { 
   
        static void Main(string[] args) 
   
        { 
   
            var container = EAS.Objects.ContainerBuilder.BuilderDefault(); 
   
            var bus = container.GetComponentInstance("MessageBus") as IMessageBus; 
   
            System.Console.WriteLine("Subscriber"); 
   
   
   
            bus.Subscribe(new Program(), "Subscriber1", Demo.Messages.Topics.DEMO_TOPIC, MessageNotify); 
   
            System.Console.ReadLine(); 
   
        } 
   
   
   
        static void MessageNotify(object m) 
   
        { 
   
            Demo.Messages.Message message = m as Demo.Messages.Message; 
   
            System.Console.WriteLine(string.Format("Subscribe:{0}", message.ID)); 
   
        } 
   
    } 
   
}

其中代碼bus.Subscribe(new Program(), "Subscriber1", Demo.Messages.Topics.DEMO_TOPIC, MessageNotify);：完成把消息訂閱到MessageNotify通知函數之中。

在Demo.Server項目之中，啓動服務器而且開始接收訂閱和發佈：

 
using System; 
   
using System.Collections.Generic; 
   
using System.Linq; 
   
using System.Text; 
   
using EAS.Sockets; 
   
   
   
namespace Demo.Server 
   
{ 
   
    class Program 
   
    { 
   
        static void Main(string[] args) 
   
        { 
   
            SocketServer socketServer = new SocketServer(128); 
   
            socketServer.Port = 6606; 
   
            socketServer.Logger = new EAS.Loggers.ConsoleLogger();  
   
            socketServer.Initialize(); 
   
            System.Console.WriteLine("Server Starting..."); 
   
            socketServer.StartServer(); 
   
            System.Console.WriteLine("Server Startup!"); 
   
            System.Console.ReadLine(); 
   
        } 
   
    } 
   
}

到此爲止，全部代碼均已完成，是否是很簡單，接下來，咱們跑起來驗證一下效果。

4、驗證效果

咱們在編譯輸入目錄Publish下先啓動Demo.Server.exe，再啓動兩個Demo.Subscriber.exe,再啓動一個Demo.Publisher.exe，在Demo.Publisher.exe控制檯按回車鍵：

OK，搞定。

5、源代碼下載

本程序的源代碼已上傳到服務器，請經過http://112.74.65.50/downloads/eas/Demo.Pub_Sub.rar進行下載，若是在開發過程之中想要了解更多有關Socket通訊框架以及更多AgileEAS.NET SOA中間件平臺的技術資源，請經過AgileEAS.NET SOA 網站：http://www.smarteas.net的最新下載欄目進行下載。