分享一个分布式消息总线，基于.NET Socket Tcp的发布-订阅框架，附代码下载

浏览数：42 / 时间：2015年06月09日

一、分布式消息总线

在很多MIS项目之中都有这样的需求，需要一个及时、高效的的通知机制，即比如当使用者A完成了任务X，就需要立即告知使用者B任务X已经完成，在通常的情况下，开发人中都是在使用者B所使用的程序之中写数据库轮循代码，这样就会产品一个很严重的两个问题，第一个问题是延迟，轮循机制要定时执行，必须会引起延迟，第二个问题是数据库压力过大，当进行高频度的轮循会生产大量的数据库查询，并且如果有大量的使用者进行轮循，那数据库的压力就更大了。

那么在这个时间，就需要一套能支持发布-订阅模式的分布式消息总线，那这个问题就可以很好的解决了，比如采用一些成熟的消息总线进行实现，比如MSMQ或者采用比如开源的NServiceBus的发布订阅机制就可以实现处理这种需求，其系统结构就会变成如下所示：

本分布式消息总线，目前广泛的被应用于分布式缓存的更新通知，当在N百台客户短在使用缓存的过程之中，某个操作修改了缓存的数据，必须会导致其他终端缓存的失效，那么使用基于Socket的分布式消息总线之后，我们可以做了修改了即可实时通知，做到缓存数据保持最新，再比如医疗应用之中的危机机管理，当发现检验、检查危急值之后，需要及时通知病人做出相应的措施，启动声光报警系统等，再比如应用于异构系统整合，当检验系统做出检验报告，通过消息总线进行发布，HIS系统则即时会收到检验报告数据而实现系统的整合。

二、基于Socket的实现

目前能够实现发布订阅模式的开源产品非常之多，为什么还要制造轮子呢，其主要原因有以下几点

1）像NServiceBus这种东西基于MSMQ，在大量的发布者-订阅者的情况下性能不佳。

2）此类东西太过于庞大、不易使用和配置。

3）学习成本过高。

那为什么要使用Socket技术进行实现呢，其主要原因是有以下几点:

1）使用高效的Socket通信技术，高效、支持更多的客户端。

2）使用简单，不需要定义太多额外的东西，只需要定义主题和消息即可使用。

目前本发布订阅框架是基于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的订阅方法进行订阅，当你订阅了某个主题之后，有发布者发布该主题的消息，你即可以收到消息并调用订阅回调函数进行处理。

三、实现一个简单的例子

现在我们开始一个简单的应用消息总线的例子，本例子代码解决方案由下图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();

        }

    }

}

到此为止，所有代码均已完成，是不是很简单，接下来，我们跑起来验证一下效果。

四、验证效果

我们在编译输入目录Publish下先启动Demo.Server.exe，再启动两个Demo.Subscriber.exe,再启动一个Demo.Publisher.exe，在Demo.Publisher.exe控制台按回车键：

OK，搞定。

五、源代码下载

本程序的源代码已上传到服务器，请通过http://42.121.30.77/downloads/eas/Demo.Pub_Sub.rar进行下载，如果在开发过程之中想要了解更多有关Socket通信框架以及更多AgileEAS.NET SOA中间件平台的技术资源，请通过AgileEAS.NET SOA 网站：http://www.smarteas.net的最新下载栏目进行下载。