分享一个分布式消息总线,基于.NET Socket Tcp的发布-订阅框架,附代码下载
2014-08-14 08:55
911 查看
一、分布式消息总线
在很多MIS项目之中都有这样的需求,需要一个及时、高效的的通知机制,即比如当使用者A完成了任务X,就需要立即告知使用者B任务X已经完成,在通常的情况下,开发人中都是在使用者B所使用的程序之中写数据库轮循代码,这样就会产品一个很严重的两个问题,第一个问题是延迟,轮循机制要定时执行,必须会引起延迟,第二个问题是数据库压力过大,当进行高频度的轮循会生产大量的数据库查询,并且如果有大量的使用者进行轮循,那数据库的压力就更大了。那么在这个时间,就需要一套能支持发布-订阅模式的分布式消息总线,那这个问题就可以很好的解决了,比如采用一些成熟的消息总线进行实现,比如MSMQ或者采用比如开源的NServiceBus的发布订阅机制就可以实现处理这种需求,其系统结构就会变成如下所示:
![](https://oscdn.geek-share.com/Uploads/Images/Content/202003/15/d488cbb7e2eb1061eb0f32be083ab540.gif)
本分布式消息总线,目前广泛的被应用于分布式缓存的更新通知,当在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个项目组成:![](https://oscdn.geek-share.com/Uploads/Images/Content/202003/15/af747e5c2a2b9050fc8b251d9d4efa92.jpg)
其中: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控制台按回车键:![](https://oscdn.geek-share.com/Uploads/Images/Content/202003/15/30af411d70b8f2fbb8cac0ba5da44d2f.png)
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人
邮件:james@agilelab.cn,mail.james@qq.com,
电话:18629261335。
相关文章推荐
- 分享一个分布式消息总线,基于.NET Socket Tcp的发布-订阅框架,附代码下载 推荐
- 分布式消息总线,基于.NET Socket Tcp的发布-订阅框架之离线支持,附代码下载
- 基于.NET Socket Tcp的发布-订阅框架
- java分布式开发TCP/IP NIO无阻塞 Socket((基于消息方式实现系统间的通信) )(转)
- 分享在Linux下使用OSGi.NET插件框架快速实现一个分布式服务集群的方法
- 一个基于反射的消息发布订阅模型详解
- 分享一个基于jQuery,backbone.js和underscore.js的消息提示框架 - Backbone.Notifier
- 分享一个基于jQuery,backbone.js和underscore.js的消息提示框架 -...
- Kafka是一种分布式的,基于发布/订阅的消息系统
- Kafka 分布式的,基于发布/订阅的消息系统
- 基于WCF和MSMQ构建发布/订阅消息总线(Pub/Sub Message Bus) 推荐
- 分享基于分布式Http长连接框架--代码模型
- 分享一个基于jQuery,backbone.js和underscore.js的消息提示框架 - Backbone.Notifier
- SSB Router 发布、订阅框架PPT 文档 – 分享下载!
- 分布式消息总线订阅框架
- 一个基于VB.net的异步Socket网络TCP通信可防止任意一端意外终止TCP连接的类,。
- 基于WCF和MSMQ构建发布/订阅消息总线(Pub/Sub Message Bus)
- Asp.net 2.0 自定义控件开发[开发一个图表(WebChart)控件(柱状图示例)](示例代码下载)
- JavaScript组件之JQuery(A~Z)教程(基于Asp.net运行环境)[示例代码下载]
- JavaScript组件之JQuery(A~Z)教程(基于Asp.net运行环境)[示例代码下载]