首页 > 代码库 > SignalR循序渐进（三）简易的集群通讯组件
SignalR循序渐进（三）简易的集群通讯组件
上一篇演示了泛型Hub的实现，微软于6月17日更新了SignalR 2.1.0，然后自带了泛型Hub，于是就不需要自己去实现了…（微软你为啥不早一个月自带啊…）。不过没关系，SignalR出彩之处不在泛型Hub，本篇为各位观众带来了基于SignalR的简易集群通讯组件Demo，可用于分布式定时任务。
说到集群，自然想到了NLB啊Cluster啊HPC啊等等。NLB受制于成员数量，Cluster用数量堆高可用性，HPC太复杂。本着SignalR的双向异步通讯的特点，其实是可以用来玩弹性计算的。初始状态由一台计算任务分发节点，一台监控以及一台计算节点构成。随着任务分发队列中的任务数越来越多，一台执行节点无法及时消耗待执行任务，达到某个阈值的时候，动态的加入一个计算节点来增加计算吞吐量。同样的，当队列中的任务基本处于很低的数量的时候，自动移除一个计算节点来减少资源消耗。当然，如果是大型的计算量之下，分发节点，队列都应该是集群的，还要考虑各种计算节点故障之类的问题，这不在本篇考虑的范畴内，本篇以初始状态模型来一步步实现简易集群通讯组件。
好，废话不说了，正篇开始。
任务分发节点
任务分发节点只有一个公开的行为，就是接受计算节点任务执行完成的消息。
下面是实现。
/// <summary> /// 集群交换器 /// </summary> public class ClusterHub : Hub<IClusterClient> { /// <summary> /// /// </summary> static ClusterHub() { aliveDictionary = new ConcurrentDictionary<string, Guid>(); } /// <summary> /// /// </summary> /// <param name="dispatcher"></param> public ClusterHub(IDispatcher dispatcher) { this.dispatcher = dispatcher; db = OdbFactory.Open(localDbFileName); } /// <summary> /// 本地数据库文件名 /// </summary> const string localDbFileName = "ClusterStorage.dll"; /// <summary> /// 监视器连接Id /// </summary> static string monitorConnectionId; /// <summary> /// 调度器 /// </summary> IDispatcher dispatcher; /// <summary> /// 在线词典 /// </summary> static ConcurrentDictionary<string, Guid> aliveDictionary; /// <summary> /// /// </summary> static IOdb db; /// <summary> /// 完成任务 /// </summary> /// <param name="jobResult"></param> public void Finished(Contracts.Messages.JobResultDto jobResult) { lock (db) { var members = db.AsQueryable<MemberDo>(); var member = members.SingleOrDefault(m => m.Id == Guid.Parse(jobResult.Id)); if (member != null) { member.UpdateStatisticsInfo(jobResult.ProcessedTime); db.Store(member); if (!string.IsNullOrWhiteSpace(monitorConnectionId)) { Clients.Client(monitorConnectionId).UpdateMemberStatisticsInfo(new Contracts.Messages.MemberStatisticsInfoDto() { Id = member.Id.ToString(), AverageProcessedTime = member.AverageProcessedTime }); } } } Clients.Caller.RunJob(dispatcher.GetJobId()); } /// <summary> /// 加入 /// </summary> void Join() { object ip = string.Empty; var isMonitor = Context.Request.QueryString["ClientRole"] == "Monitor"; Context.Request.Environment.TryGetValue("server.RemoteIpAddress", out ip); lock (db) { var members = db.AsQueryable<MemberDo>(); var member = members.SingleOrDefault(m => m.Ip == ip.ToString() && m.IsMonitor == isMonitor); if (member != null) { member.MemberStatusType = MemberStatusTypeEnum.Connectioned; } else { member = new MemberDo(ip.ToString(), isMonitor); if (isMonitor) { monitorConnectionId = Context.ConnectionId; } } db.Store(member); aliveDictionary.TryAdd(Context.ConnectionId, member.Id); if (!isMonitor) { if (!string.IsNullOrWhiteSpace(monitorConnectionId)) { Clients.Client(monitorConnectionId).MemberJoin(member.Id); } Clients.Caller.GetId(member.Id.ToString()); Clients.Caller.RunJob(dispatcher.GetJobId()); } } } /// <summary> /// 离开 /// </summary> void Leave() { var id = Guid.Empty; aliveDictionary.TryRemove(Context.ConnectionId, out id); lock (db) { var members = db.AsQueryable<MemberDo>(); var member = members.SingleOrDefault(m => m.Id == id); if (member != null) { member.MemberStatusType = MemberStatusTypeEnum.Disconnectioned; db.Store(member); if (member.IsMonitor) { monitorConnectionId = string.Empty; } else if (!string.IsNullOrWhiteSpace(monitorConnectionId)) { Clients.Client(monitorConnectionId).MemberLeave(id); } } } } public override Task OnConnected() { Console.WriteLine(Context.ConnectionId+":Connected"); Join(); return base.OnConnected(); } public override Task OnDisconnected() { Console.WriteLine(Context.ConnectionId + ":Disconnected"); Leave(); return base.OnDisconnected(); } public override Task OnReconnected() { Console.WriteLine(Context.ConnectionId + ":Reconnected"); return base.OnReconnected(); } }
ClusterHub承载着2种客户端角色的交互，计算节点和监控。
这边采用了一个轻量级的基于C#开发的无引擎对象数据库来存储客户端信息。
先说重载的部分：
OnConnected - 当有客户端连接的时候，执行Join方法。
OnDisconnected - 当有客户端离线的时候，执行Leave方法。
然后是私有方法：
Join - 根据QueryString来区分客户端类型是计算节点还是监视器，如果是计算节点，就直接通知监视器有成员加入，然后通过IDispatcher来获取任务Id，通知计算节点开始执行任务。
Leave - 计算节点离线的时候通知监视器。
公开方法：
Finished - 计算节点完成任务后就调用该方法，Hub将计算的一些统计信息更新到本地存储，同时通知监视器更新计算结果。
私有变量：
IDispatcher– 任务调度器接口，由外部组件来负责具体的实现。
计算节点
计算节点有两个行为：
GetId - 获取节点身份。
RunJob - 执行任务。
/// <summary> /// 集群客户端 /// </summary> public class ClusterClient { /// <summary> /// /// </summary> /// <param name="jobProvider"></param> public ClusterClient(IJobProvider jobProvider) { this.jobProvider = jobProvider; url = ConfigurationManager.AppSettings["HubAddress"]; var queryStrings = new Dictionary<string, string>(); queryStrings.Add("ClientRole", "Normal"); connection = new HubConnection(url, queryStrings); hubProxy = connection.CreateHubProxy(typeof(IClusterHub).GetCustomAttributes(typeof(DescriptionAttribute), false).OfType<DescriptionAttribute>().First().Description); InitClientEvents(); connection.Start().Wait(); } string url; HubConnection connection; IHubProxy hubProxy; IJobProvider jobProvider; string id; /// <summary> /// /// </summary> void InitClientEvents() { hubProxy.On("GetId", (id) => GetId(id)); hubProxy.On("RunJob", (jobId) => RunJob(jobId)); } /// <summary> /// 执行任务 /// </summary> /// <param name="id"></param> void GetId(string id) { this.id = id; } /// <summary> /// 执行任务 /// </summary> /// <param name="jobId"></param> void RunJob(string jobId) { var startTime = DateTime.Now; jobProvider.Invoke(jobId); var stopTime = DateTime.Now; hubProxy.Invoke("Finished", new JobResultDto() { Id = id, JobId = jobId, ProcessedTime = (stopTime - startTime).TotalMilliseconds }); } }
客户端的实现很简单，核心就是通过构造函数注入任务提供接口，由接口通过任务Id来执行任务。
监视器
监视器具有三个公开行为：
MemberJoin - 计算节点加入
MemberLeave - 计算节点离线
UpdateMemberStatisticsInfo - 更新节点统计信息
/// <reference path="jquery-2.1.1.js" />/// <reference path="jquery.signalR-2.1.0.js" />(function ($) { var members = []; var methods = { reloadList: function () { var list = ""; $.each(members, function (i, n) { list += "<li id=‘member_" + n.Id + "‘>[" + n.Id + "]:AverageProcessedTime " + n.AverageProcessedTime + " Milliseconds</li>"; }); $(‘#members‘).html(list); } } var hubs = { clusterHub: $.connection.clusterHub, init: function () { $.connection.hub.logging = true; $.connection.hub.url = ‘http://192.168.1.124:10086/signalr‘; $.connection.hub.qs = { "ClientRole": "Monitor" } $.connection.hub.start().done(function () { }); } } var cluster = { on: { updateMemberStatisticsInfo: function (data) { $.each(members, function (i, n) { if (n.Id == data.Id) { n.AverageProcessedTime = data.AverageProcessedTime; return; } }); methods.reloadList(); }, memberJoin: function (id) { members.push({ "Id": id, "AverageProcessedTime": 0 }); methods.reloadList(); }, memberLeave: function (id) { members = $.grep(members, function (n) { return n.Id != id }); methods.reloadList(); } } } $(function () { hubs.clusterHub.client.UpdateMemberStatisticsInfo = cluster.on.updateMemberStatisticsInfo; hubs.clusterHub.client.MemberJoin = cluster.on.memberJoin; hubs.clusterHub.client.MemberLeave = cluster.on.memberLeave; hubs.init(); });})(jQuery);
<!DOCTYPE html><html xmlns="http://www.w3.org/1999/xhtml"><head> <title>集群监视器</title></head><body> <div> <ul id="members"></ul> </div> <script src="scripts/jquery-2.1.1.min.js"></script> <script src="scripts/jquery.signalR-2.1.0.min.js"></script> <script src="http://192.168.1.124:10086/signalr/hubs"></script> <script src="scripts/core.js"></script></body></html>
监视器用real-time的Web平台实现，一共注册三个方法的实现。
最终效果
Hub端启动后，先启动监视器，然后在不同的机器上启动计算端，图上是2个计算节点，监视器上也显示着2个节点，每个节点执行一个JobId后，监视器上就会刷新结果。
进一步思考和扩展
简易集群组件就到这儿了，本篇演示的是一个思路，可以在这个基础上深度扩展成文章开头所描述的那样，高性能高可用的基于SignalR的集群组件。欢迎各位有兴趣的同学进行讨论和拍砖。
转载请注明出处：http://www.cnblogs.com/royding/p/3811169.html
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。