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

一、分布式消息总线

在很多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：

   1: using System;
   2: using System.Collections.Generic;
   3: using System.Linq;
   4: using System.Text;
   5: using System.Collections;
   6:  
   7: namespace EAS.Messages
   8: {
   9:     /// <summary>
  10:     /// 消息总线接口定义。
  11:     /// </summary>
  12:     public interface IMessageBus : IDisposable
  13:     {
  14:         /// <summary>
  15:         /// 注册发布者。
  16:         /// </summary>
  17:         /// <param name="publisher">发布者。</param>
  18:         void AddPublisher(string publisher);
  19:  
  20:         /// <summary>
  21:         /// 注册发布者。
  22:         /// </summary>
  23:         /// <param name="publisher">发布者。</param>
  24:         /// <param name="topic">主题。</param>
  25:         void AddPublisher(string publisher, string topic);
  26:  
  27:         /// <summary>
  28:         /// 发布一条消息到总线。
  29:         /// </summary>
  30:         /// <param name="topic">主题。</param>
  31:         /// <param name="message">发布的消息。</param>
  32:         void Publish(string topic, object message);
  33:  
  34:         /// <summary>
  35:         /// 订阅消息。
  36:         /// </summary>
  37:         /// <param name="subscriber">订阅者。</param>
  38:         /// <param name="topic">主题。</param>
  39:         /// <param name="notifyHandler">订阅通知。</param>
  40:         void Subscribe(object subscriber, string topic, MessageNotifyHandler notifyHandler);
  41:  
  42:         /// <summary>
  43:         /// 订阅消息。
  44:         /// </summary>
  45:         /// <param name="subscriber">订阅者。</param>
  46:         /// <param name="friendName">订阅者名称，用于处理离线订阅。</param>
  47:         /// <param name="topic">主题。</param>
  48:         /// <param name="notifyHandler">订阅通知。</param>
  49:         void Subscribe(object subscriber,string friendName ,string topic, MessageNotifyHandler notifyHandler);
  50:  
  51:         /// <summary>
  52:         /// 订阅消息。
  53:         /// </summary>
  54:         /// <param name="subscriber">订阅者。</param>
  55:         /// <param name="friendName">订阅者名称，用于处理离线订阅。</param>
  56:         /// <param name="topic">主题。</param>
  57:         /// <param name="notifyHandler">订阅通知。</param>
  58:         /// <param name="changedHandler">发布者状态变化委托。</param>
  59:         void Subscribe(object subscriber, string friendName, string topic, MessageNotifyHandler notifyHandler,PublisherSstatusChangedHandler changedHandler);
  60:  
  61:         /// <summary>
  62:         /// 退订消息。
  63:         /// </summary>
  64:         /// <param name="subscriber">订阅者。</param>
  65:         void Unsubscribe(object subscriber);
  66:  
  67:         /// <summary>
  68:         /// 退订消息。
  69:         /// </summary>
  70:         /// <param name="subscriber">订阅者。</param>
  71:         /// <param name="topic">主题。</param>
  72:         void Unsubscribe(object subscriber, string topic);
  73:  
  74:         /// <summary>
  75:         /// 退订消息。
  76:         /// </summary>
  77:         /// <param name="subscriber">订阅者。</param>
  78:         /// <param name="friendName">订阅者名称，用于处理离线订阅。</param>
  79:         /// <param name="topic">主题。</param>
  80:         void Unsubscribe(object subscriber, string friendName, string topic);
  81:     }
  82: }

IMessageBus就基于Socket发布订阅消息总线的灵魂接口，也是基唯一的发布者调用者功能入口，也就是说不管你是发布者还是订阅者都需要调用这个接口，如果你是发布者请调用IMessageBus接口的Publish方法向消息总线发布消息，如果是你订阅者请通过IMessageBus的订阅方法进行订阅，当你订阅了某个主题之后，有发布者发布该主题的消息，你即可以收到消息并调用订阅回调函数进行处理。

三、实现一个简单的例子

现在我们开始一个简单的应用消息总线的例子，本例子代码解决方案由下图4个项目组成：

其中：Demo.Messages项目定义发布者、订阅者所使用的消息对象和消息主题。

Demo.Publisher项目为发布者代码。

Demo.Subscriber项目为订阅者代码。

Demo.Server项目为服务端代码。

在Demo.Messages项目之中，我们定义了一个消息Message：

   1: using System;
   2: using System.Collections.Generic;
   3: using System.Linq;
   4: using System.Text;
   5: using System.Xml.Serialization;
   6:  
   7: namespace Demo.Messages
   8: {
   9:     [Serializable]
  10:     public class Message
  11:     {
  12:         [XmlElement]
  13:         public Guid ID
  14:         {
  15:             get;
  16:             set;
  17:         }
  18:     }
  19: }

消息Message很简单，只有一个属性ID，同时 还需要定义一个消息主题：

   1: using System;
   2: using System.Collections.Generic;
   3: using System.Linq;
   4: using System.Text;
   5:  
   6: namespace Demo.Messages
   7: {
   8:     public class Topics
   9:     {
  10:         public static readonly string DEMO_TOPIC = "演示消息";
  11:     }
  12: }

我们定义了一个消息主题为“演示消息”。

在Demo.Publisher项目之中，没有太多额外的代码，只有在Program.cs写了以下简单的调用代码：

   1: using System;
   2: using System.Collections.Generic;
   3: using System.Linq;
   4: using System.Text;
   5: using EAS.Messages;
   6:  
   7: namespace Demo.Publisher
   8: {
   9:     class Program
  10:     {
  11:         static void Main(string[] args)
  12:         {
  13:             var container = EAS.Objects.ContainerBuilder.BuilderDefault();
  14:             var bus = container.GetComponentInstance("MessageBus") as IMessageBus;
  15:             System.Console.WriteLine("Publisher");
  16:  
  17:             while (System.Console.ReadLine()!="exit")
  18:             {
  19:                 var m = new Messages.Message { ID = Guid.NewGuid() };
  20:                 bus.Publish(Demo.Messages.Topics.DEMO_TOPIC, m);
  21:                 System.Console.WriteLine(string.Format("Publish:{0}", m.ID));
  22:             }
  23:         }
  24:     }
  25: }

从IOC容器获取一个消息总线IMessageBus对象，并调用Publish函数发布消息”bus.Publish(Demo.Messages.Topics.DEMO_TOPIC, m);“。

当然了，使用了IOC容器，就离来开配置文件了，其App.config文件内容如下：

   1: <?xml version="1.0" encoding="utf-8"?>
   2: <configuration>
   3:   <configSections>
   4:     <section name="eas" type="EAS.ConfigHandler,EAS.MicroKernel" />
   5:   </configSections>
   6:   <eas>
   7:     <objects>
   8:       <!--消息总线-->
   9:       <object name="MessageBus" assembly="EAS.MicroKernel" type="EAS.Sockets.Bus.SocketBus" LifestyleType="Singleton">
  10:         <property name="Url" type="string" value="socket.tcp://127.0.0.1:6606/"/>
  11:       </object>
  12:     </objects>
  13:   </eas>
  14: </configuration>

在Demo.Subscriber项目之中，使用与Demo.Publisher一模一样的配置文件，其Program.cs代码如下：

   1: using System;
   2: using System.Collections.Generic;
   3: using System.Linq;
   4: using System.Text;
   5: using EAS.Messages;
   6:  
   7: namespace Demo.Subscriber
   8: {
   9:     class Program
  10:     {
  11:         static void Main(string[] args)
  12:         {
  13:             var container = EAS.Objects.ContainerBuilder.BuilderDefault();
  14:             var bus = container.GetComponentInstance("MessageBus") as IMessageBus;
  15:             System.Console.WriteLine("Subscriber");
  16:  
  17:             bus.Subscribe(new Program(), "Subscriber1", Demo.Messages.Topics.DEMO_TOPIC, MessageNotify);
  18:             System.Console.ReadLine();
  19:         }
  20:  
  21:         static void MessageNotify(object m)
  22:         {
  23:             Demo.Messages.Message message = m as Demo.Messages.Message;
  24:             System.Console.WriteLine(string.Format("Subscribe:{0}", message.ID));
  25:         }
  26:     }
  27: }

其中代码bus.Subscribe(new Program(), "Subscriber1", Demo.Messages.Topics.DEMO_TOPIC, MessageNotify);：完成把消息订阅到MessageNotify通知函数之中。

在Demo.Server项目之中，启动服务器并且开始接收订阅和发布：

   1: using System;
   2: using System.Collections.Generic;
   3: using System.Linq;
   4: using System.Text;
   5: using EAS.Sockets;
   6:  
   7: namespace Demo.Server
   8: {
   9:     class Program
  10:     {
  11:         static void Main(string[] args)
  12:         {
  13:             SocketServer socketServer = new SocketServer(128);
  14:             socketServer.Port = 6606;
  15:             socketServer.Logger = new EAS.Loggers.ConsoleLogger(); 
  16:             socketServer.Initialize();
  17:             System.Console.WriteLine("Server Starting...");
  18:             socketServer.StartServer();
  19:             System.Console.WriteLine("Server Startup!");
  20:             System.Console.ReadLine();
  21:         }
  22:     }
  23: }

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

四、验证效果

我们在编译输入目录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的最新下载栏目进行下载。

六、问题反馈

麻烦大家在通过视频进行学习的时候能及时把问题反馈给楼主，或者有什么需要改进的一些建议都请向楼主直接反馈,以下是联系方式：

AgileEAS.NET SOA 网站：http://www.smarteas.net

官方博客:http://eastjade.cnblogs.com

楼主QQ:47920381,AgileEAS.NET

QQ群：113723486（AgileEAS SOA 平台）/上限1000人

199463175（AgileEAS SOA 交流）/上限1000人

120661978（AgileEAS.NET 平台交流）/上限1000人

邮件:[email protected],[email protected],

电话：18629261335。

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

时间： 2024-10-01 14:58:33

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

分布式消息总线，基于.NET Socket Tcp的发布-订阅框架之离线支持，附代码下载

一.分布式消息总线以及基于Socket的实现 在前面的分享一个分布式消息总线,基于.NET Socket Tcp的发布-订阅框架,附代码下载一文之中给大家分享和介绍了一个极其简单也非常容易上的基于.NET Socket Tcp 技术实现的分布消息总线,也是一个简单的发布订阅框架: 并且以案例的形式为大家演示了如何使用这个分布式消息总线架构发布订阅架构模式的应用程序,在得到各位同仁的反馈的同时,大家也非常想了解订阅者离线的情况,即支持离线构发布订阅框架. 二.离线架构 不同于订阅者.发布者都同时在

分布式消息总线

消息总线是一种通信工具,可以在机器之间互相传输消息.文件等. 消息总线扮演着一种消息路由的角色,拥有一套完备的路由机制来决定消息传输方向.发送段只需要向消息总线发出消息而不用管消息被如何转发,为了避免消息丢失,部分消息总线提供了一定的持久化存储和灾备的机制. 分布式消息总线比较 开源消息总线ActiveMQ

分享一个自己用的基于mvc编程工作管理

前言: 最近在家没事学习下mvc,正好把以前用webform写的一个帮助自己编码的工具重构成了mvc,另外根据自己的编程工作感悟添加了公司常用软件维护 ,数据库操作记录这些新功能. 技术没什么高深的技术,就是mvc+jquery easyUi+简单的三层,生成文档的模板引擎用的Razor. 主要还是解决沟通成本太高的问题,都是根据工作中遇到的问题慢慢解决的,至少很多功能可以解决小型软件团队很多流程问题,自己摸索真是又痛苦又快乐.相信也有很多朋友和我一样没有去过大型软件公司,很多问题只能看别人的文

分享一个从国外网站看到的socket程序，程序并不强悍,但能从代码中看到老外是怎么写代码的,咱可以取长补短

原文:分享一个从国外网站看到的socket程序,程序并不强悍,但能从代码中看到老外是怎么写代码的,咱可以取长补短 源代码下载地址:http://www.zuidaima.com/share/1550463702764544.htm package com.zuidaima.xs; /**************************************************************** * Version : 1.0 * Date : 02/03/2007 * Autho

分享一个自定义的console类 让你不再纠结JS中的调试代码的兼容

分享一个自定义的console类 让你不再纠结JS中的调试代码的兼容 在写JS的过程中,为了调试我们常常会 写很多 console.log.console.info.console.group.console.warn.console.error代码来查看JS 的运行情况,但发布时又因为IE不支持console,又要去掉这些代码,一不小心就会出错 问题的产生 在写JS的过程中,为了调试我们常常会写很多 console.log.console.info.console.group.console.

kafka：一个分布式消息系统

1.背景 最近因为工作需要,调研了追求高吞吐的轻量级消息系统Kafka,打算替换掉线上运行的ActiveMQ,主要是因为明年的预算日流量有十亿,而ActiveMQ的分布式实现的很奇怪,所以希望找一个适合分布式的消息系统. 以下是内容是调研过程中总结的一些知识和经验,欢迎拍砖. 2.基础知识 2.1.什么是消息队列 首先,我们来看看什么是消息队列,维基百科里的解释翻译过来如下: 队列提供了一种异步通信协议,这意味着消息的发送者和接收者不需要同时与消息保持联系,发送者发送的消息会存储在队列中,直到接

分享一个自己写的基于TP的关系模型(四)

修复分页BUG,原有代码查询到最后一页就一只查询最后一页 $this->nowPage = $this->nowPage>0 ? $this->nowPage : 1; $this->maxPage = ceil($this->total/$this->rows); //$this->nowPage > $this->maxPage and $this->nowPage = $this->maxPage; $this->firs

分享一个点赞超过100的漂亮ASP.NET MVC蓝色界面框架

从 陈贞宝 博客中看到一个MVC模板感觉特别漂亮就尝试着分离出来,直接拿来用啦,直接拷贝到自己的常用的代码库里收藏起来,地址是http://www.cnblogs.com/baihmpgy/p/3817613.html 下图是该界面框架的样式. 要使用该界面框架,你可以通过以下地址来下载到界面框架的Visual Studio 2013模板. 模板下载地址:http://download.csdn.net/detail/luomingui/8297843 下载后(不解压缩直接拷过去也可以),解压缩

基于OpenGL的随机图动画演示框架与代码

基于Opengl的实时动画,为可视化科学计算过程提供了有利的工具. 然而,对于初学者来说,相关知识的学习仍需经历一个过程. 将Opengl的初始化等进行适当封装,建立了Opengl的动画框架,用户可以直接应用C语言的知识或Matlab知识,定义框架中相应的函数,即可实现实时动画程序开发,为科学计算和可视化计算提供了便利. 下面给出了Ubuntu14.04下,基于OpenGL的点线随机图动画框架. 通过分析该程序,用户可以方便地定制程序,以达到用户算法的期望. 代码如下: //第三方库: sudo