“一切都是消息”--MSF(消息服务框架)之【发布-订阅】模式
2017-10-11 15:18
453 查看
一、定时推送模式
这是最普通最常见的推送模式,只要客户端订阅了MSF的服务,服务器会每隔一秒向客户端推送一次服务处理结果。在下面的示例中,我们先来演示一个简单的“服务器时间服务”的功能。
1.1,在TestService项目添加一个类文件 TimeService.cs ,其代码如下:
using System; using System.Collections.Generic; using System.Linq; using System.Text; namespace TestService { public class TimeService:ServiceBase { public DateTime ServerTime() { return DateTime.Now; } } }
注意:今天我们这个MSF服务类TimeService 集成的不是前一篇说的IService接口,而是 ServiceBase 抽象类,实际上它也是实现了IService接口的类,这样可以让我们的服务类代码更简单。
别忘了,在IOC配置文件 IOCConfig.xml 注册我们新添加的服务:
<IOC Name="TestService"> <Add Key="TestTimeService" InterfaceName="IService" FullClassName="TestService.TimeService" Assembly="TestService" /> <!-- 其它略 --> </IOC>
该配置需要注意3点:
虽然TimeService 继承的是ServiceBase 对象,但在这里配置 InterfaceName的时候,仍然使用 IService
Key="TestTimeService" 而不是 Key="TimeService" ,实际上这里配置的Key 可以是任意名字,只要跟配置文件中其它Key的值不重复即可
调用服务的时候,ServiceRequest 对象的 ServiceName 属性指定的服务名称,是这里配置的Key的值,而不是MSF服务类的类名
1.2,在TestClient 项目添加订阅服务的代码:
在订阅前,我们可以直接请求下上面的【服务器时间】服务,测试下服务是否可行:DateTime serverTime = client.RequestServiceAsync<DateTime>("Service://TestTimeService/ServerTime/", PWMIS.EnterpriseFramework.Common.DataType.DateTime).Result; Console.WriteLine("MSF Get Server Time:{0}", serverTime);
测试成功,下面继续编写订阅模式的代码:
ServiceRequest request3 = new ServiceRequest(); request3.ServiceName = "TestTimeService"; request3.MethodName = "ServerTime"; int count = 0; client.Subscribe<DateTime>(request3, PWMIS.EnterpriseFramework.Common.DataType.DateTime, s => { if (s.Succeed) { Console.WriteLine("MSF Server Time:{0}", s.Result); } else { Console.WriteLine("MSF Server Error:{0}", s.ErrorMessage); } count++; if (count > 10) { client.Close(); Console.WriteLine("订阅【服务器时钟服务】结束。按回车键继续。"); } });
与请求模式不同,客户端要使用订阅模式,只需要将服务代理类的 RequestService 方法替换成 Subscribe 方法,该方法的第一个泛型参数类型表示订阅的结果的类型。
由于是订阅模式, Subscribe 不提供Async的同名方法,因为服务器会多次向客户端推送订阅的结果,何时订阅结束,可以由客户端来决定,在客户端提供的服务端回调方法内来关闭订阅的连接即可。所以Subscribe 方法的下一行代码会立即执行,无法实现RequestServiceAsync 这种“同步”效果。
在当前示例中,服务端会向客户端推送10次服务器时间,然后客户端会关闭订阅连接。假如客户端不关闭订阅连接,服务器会一直向客户端推送订阅结果,每秒推送一次。
下面是这个示例的运行结果:
MSF Server Time:2017-10-11 10:33:48 MSF Server Time:2017-10-11 10:33:49 MSF Server Time:2017-10-11 10:33:50 MSF Server Time:2017-10-11 10:33:51 MSF Server Time:2017-10-11 10:33:52 MSF Server Time:2017-10-11 10:33:53 MSF Server Time:2017-10-11 10:33:54 MSF Server Time:2017-10-11 10:33:55 MSF Server Time:2017-10-11 10:33:56 MSF Server Time:2017-10-11 10:33:58 MSF Server Time:2017-10-11 10:33:59 订阅【服务器时钟服务】结束。按回车键继续。
二、事件推送模式
有时候我们并不需要固定间隔时间(例如每秒)调用服务方法然后将处理结果推送给客户端,而是在某个特定的时间才向客户端推送订阅的服务结果,这个需求可以在服务端实现一个定时器,在时间到了后才推送,或者,进行某项业务处理过程,满足某项业务条件后,触发一个业务事件,在这个业务事件中,将订阅的结果推送给客户端。定时器处理的是它触发的事件,业务处理过程也可以触发某种业务操作事件,所以这种推送模式,就是“事件推送模式”,跟前面的“定时推送模式”是完全不同的模式,在事件推送模式中,看起来是将服务端的事件,推送到客户端订阅的方法里面去了,事件的实际处理,到了客户端,因此,事件推送模式,也是一种“分布式事件”处理模式。
下面我们来实现一个“闹铃服务”,客户端订阅此闹铃服务,指定响铃的时间和响铃的次数,服务端的闹铃到了指定时间,就会向客户端推送“闹铃服务”:“闹铃响了”,一直推送到客户端指定的次数为止。
2.1,在TestService项目添加闹钟服务类文件 AlarmClockService.cs,其代码如下:
public class AlarmClockService:ServiceBase { System.Timers.Timer timer; DateTime AlarmTime; int AlarmCount; int MaxAlarmCount; public event EventHandler Alarming; public AlarmClockService() { timer = new System.Timers.Timer(); timer.Interval = 10000; timer.Elapsed += timer_Elapsed; } void timer_Elapsed(object sender, System.Timers.ElapsedEventArgs e) { if (e.SignalTime >= this.AlarmTime) { if (Alarming != null) Alarming(this, new EventArgs()); base.CurrentContext.PublishData(DateTime.Now); //e.SignalTime AlarmCount++; Console.WriteLine("AlarmClockService Publish Count:{0}", AlarmCount); } else { Console.WriteLine("Alarm Time:{0},AlarmClock waiting...",this.AlarmTime); } if (AlarmCount > MaxAlarmCount) { timer.Stop(); //推送一个结束标记值:1900-1-1 base.CurrentContext.PublishData(new DateTime(1900, 1, 1)); Console.WriteLine("[{0}] AlarmClockService Timer Stoped. ", new DateTime(1900,1,1)); base.CurrentContext.PublishEventSource.DeActive(); } } public ServiceEventSource SetAlarmTime(AlarmClockParameter para) { this.MaxAlarmCount = para.AlarmCount; this.AlarmTime = para.AlarmTime; return new ServiceEventSource(timer, 2, () => { //要初始化执行的代码或者方法 AlarmCount = 0; timer.Start(); //如果上面的代码是一个执行时间比较长的方法,但又不知道何时执行完成, //并且不想等待超时回收服务对象,而是在执行完成后立即回收服务对象,可以调用下面的代码: //CurrentContext.PublishEventSource.DeActive(); //注意:调用DeActive 方法后将会停止事件推送,所以请注意此方法调用的时机。 //下面代码仅做测试,查看服务事件源对象的活动生命周期 //在 ActiveLife 时间之后,一直没有事件推送,则事件源对象被视为非活动状态,发布工作线程会被回收。 //在本例中,ActiveLife 为ServiceEventSource 构造函数的第二个参数,值为 2分钟,可以通过下面一行代码证实: int life = base.CurrentContext.PublishEventSource.ActiveLife; //如果上面执行的是一个执行时间比较长的方法,并且有返回值,想将返回值也推送给订阅端,可以再次执行CurrentContext.PublishData //CurrentContext.PublishData(DateTime.Now); //如果事件推送结束,需要设置事件源为非活动状态,否则,需要等待 ActiveLife 时间之后自然过期成为非活动状态。 //如果你无法确定事件推送何时结束,请不要调用下面的方法 //CurrentContext.PublishEventSource.DeActive(); }); } }
相关文章推荐
- “一切都是消息”--MSF(消息服务框架)之【发布-订阅】模式
- “一切都是消息”--MSF(消息服务框架)之【发布-订阅】模式
- “一切都是消息”--MSF(消息服务框架)之【请求-响应】模式(点对点)
- “一切都是消息”--MSF(消息服务框架)入门简介
- 分布式服务框架学习笔记8 ActiveMQ入门2 发布/订阅模式
- Redis基础学习--Redis 事务(watch命令)、生存时间、排序、消息通知("发布/订阅"模式)、管道、节省空间
- 发布-订阅消息模式
- RabbitMQ消息分发模式----"Publish/Subscribe"发布/订阅模式
- (三)ActiveMQ之发布- 订阅消息模式实现
- ActiveMQ发布-订阅消息模式
- 发布-订阅消息模式
- 发布-订阅消息模式
- 消息队列-ActiveMQ学习笔记(三)-发布-订阅消息模式实现
- 【转】redis 消息队列发布订阅模式spring boot实现
- Spring Data Redis实现消息队列——发布/订阅模式
- 消息队列及发布/订阅模式
- 观察者模式:消息的发布与订阅
- Redis学习笔记(十)消息通知(任务队列和发布订阅模式)
- RabbitMQ消息分发模式----"Publish/Subscribe"发布/订阅模式
- 消息队列-ActiveMQ学习笔记(三)-发布-订阅消息模式实现