在上一篇，“一切都是消息”--MSF（消息服务框架）之【请求-响应】模式，我们演示了MSF实现简单的请求-响应模式的示例，今天来看看如何实现【发布-订阅】模式。简单来说，该模式的工作过程是：

客户端发起订阅--》服务器接受订阅--》服务器处理被订阅的服务方法--》服务器将处理结果推送给客户端--》客户端收到消息--》客户端关闭订阅连接

MSF的【发布-订阅】通信模式，支持2种模式，分别是：

一、定时推送模式

这是最普通最常见的推送模式，只要客户端订阅了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
订阅【服务器时钟服务】结束。按回车键继续。

1.3，改变推送频率

默认情况下，定时推送模式是每秒推送一次，你可以在定义方法中调用基类的方法来修改它，具体代码略。

二、事件推送模式

有时候我们并不需要固定间隔时间（例如每秒）调用服务方法然后将处理结果推送给客户端，而是在某个特定的时间才向客户端推送订阅的服务结果，这个需求可以在服务端实现一个定时器，在时间到了后才推送，或者，进行某项业务处理过程，满足某项业务条件后，触发一个业务事件，在这个业务事件中，将订阅的结果推送给客户端。

定时器处理的是它触发的事件，业务处理过程也可以触发某种业务操作事件，所以这种推送模式，就是“事件推送模式”，跟前面的“定时推送模式”是完全不同的模式，在事件推送模式中，看起来是将服务端的事件，推送到客户端订阅的方法里面去了，事件的实际处理，到了客户端，因此，事件推送模式，也是一种“分布式事件”处理模式。

下面我们来实现一个“闹铃服务”，客户端订阅此闹铃服务，指定响铃的时间和响铃的次数，服务端的闹铃到了指定时间，就会向客户端推送“闹铃服务”：“闹铃响了”，一直推送到客户端指定的次数为止。

与定时推送不同的是，事件推送模式，要求被订阅的方法，返回 ServiceEventSource 类型，它表示一个事件源对象，请看下面的闹钟服务示例。

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();
            });
        }
    }

注意：

跟上面一样，不要忘记了在IOCConfig.xml文件注册此闹钟服务。

闹钟服务的类中有一个定时器对象，当订阅闹钟服务的 SetAlarmTime 方法的时候，会给闹钟服务传入必要的参数以便闹钟工作，参数类AlarmClockParameter 定义在 TestDto项目中，其代码如下：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace TestDto
{
    public class AlarmClockParameter
    {
        /// <summary>
        /// 响铃时间
        /// </summary>
        public DateTime AlarmTime { get; set; }
        /// <summary>
        /// 响铃次数
        /// </summary>
        public int AlarmCount { get; set; }
    }
}

2.2，编写闹铃服务订阅客户端

            AlarmClockParameter acp = new AlarmClockParameter();
            acp.AlarmCount = 10;
            acp.AlarmTime = alarmTime;

            ServiceRequest request4 = new ServiceRequest();
            request4.ServiceName = "AlarmClockService";
            request4.MethodName = "SetAlarmTime";
            request4.Parameters = new object[] { acp };

            client.Subscribe<DateTime>(request4,
                  PWMIS.EnterpriseFramework.Common.DataType.DateTime, 
                  s =>
                  {
                      if (s.Succeed)
                      {
                          Console.WriteLine("闹钟响了，现在时间:{0}", s.Result);
                          if (s.Result == new DateTime(1900, 1, 1))
                          {
                              client.Close();
                              Console.WriteLine("闹铃服务结束，按回车键继续。");
                          }
                      }
                      else
                      {
                          Console.WriteLine("MSF Server Error:{0}", s.ErrorMessage);
                          client.Close();
                      }
              });

这个订阅客户端，像前面订阅服务器时间一样，没有区别，这里不多解释。

2.3，注册MSF服务方法的参数类

运行此服务端和客户端，发现客户端输出了下面的异常信息：

---处理服务时错误：系统不能处理当前类型的参数：TestDto.AlarmClockParameter

这个消息是前面服务代理类的错误处理事件输出的结果：

  Proxy client = new Proxy();
   client.ErrorMessage += client_ErrorMessage;

   static void client_ErrorMessage(object sender, MessageSubscriber.MessageEventArgs e)
        {
            Console.WriteLine("---处理服务时错误：{0}",e.MessageText);
        }

现在我们去看MSF Host控制台输出的相信错误信息：

[2017-10-11 09:12:23.736]订阅消息-- From: 127.0.0.1:57822
[2017-10-11 09:12:23.752]正在处理服务请求--From: 127.0.0.1:57822,Identity:WMI2114256838
>>[PMID:1]Publish://AlarmClockService/SetAlarmTime/TestDto.AlarmClockParameter=TestDto.AlarmClockParameter
[2017-10-11 09:12:23]处理服务的时候发生异常：执行服务方法错误：
源错误信息：系统不能处理当前类型的参数：TestDto.AlarmClockParameter，
请求的Uri:
Publish://AlarmClockService/SetAlarmTime/TestDto.AlarmClockParameter=TestDto.AlarmClockParameter，
127.0.0.1:57822,WMI2114256838

错误发生时的异常对象调用堆栈：
System.ArgumentException: 系统不能处理当前类型的参数：TestDto.AlarmClockParameter
[2017-10-11 09:12:23.767]请求处理完毕(15.6339ms)--To: 127.0.0.1:57822,Identity:WMI2114256838
>>[PMID:1]消息长度：63字节 -------
result:Service_Execute_Error:系统不能处理当前类型的参数：TestDto.AlarmClockParameter
Publish Message OK.

这说明MSF服务端不识别当前调用的服务方法上的参数类型 TestDto.AlarmClockParameter ，这里需要将这个自定义的参数类型注册到MSF的IOC配置文件上：

 <IOC Name="ServiceModel">
      <Add Key="AlarmClockParameter"  InterfaceName=""  FullClassName="TestDto.AlarmClockParameter" Assembly="TestDto" />
      <!-- 其它略-->
 </IOC>

注意：服务访问需要的自定义参数类型，必须注册在 ServiceModel 节点下。

2.4，运行订阅服务

如果前面的配置都正确了，我们重新生成项目，启动MS Host 和TestClient，就可以看到客户端输出的结果了：

请输入闹铃响铃时间(示例输入格式 11:54) >>11:55
订阅闹钟服务，闹钟将在 11:55 响铃...
闹钟响了，现在时间:2017-10-11 11:55:09
闹钟响了，现在时间:2017-10-11 11:55:19
闹钟响了，现在时间:2017-10-11 11:55:29
闹钟响了，现在时间:2017-10-11 11:55:39
闹钟响了，现在时间:2017-10-11 11:55:49
闹钟响了，现在时间:2017-10-11 11:55:59
闹钟响了，现在时间:2017-10-11 11:56:09
闹钟响了，现在时间:2017-10-11 11:56:19
闹钟响了，现在时间:2017-10-11 11:56:29
闹钟响了，现在时间:2017-10-11 11:56:39
闹钟响了，现在时间:2017-10-11 11:56:49
闹钟响了，现在时间:1900-1-1 0:00:00
闹铃服务结束，按回车键继续。

在客户端控制台输入闹铃时间，我们看到在时间到了后，服务器才向客户端推送了“响铃通知”消息，客户端处理这个事件将结果打印在屏幕上。

三、MSF的Actor模式

在MSF的入门篇介绍中，我们说MSF具有实现Actor编程模型的能力，在MSF中，每一个被订阅的服务，它本质上都是一个分布式的Actor对象，这些Actor对象在第一次被订阅的时候激活，一直到没有任何客户端订阅它们为止。

对于同一个MSF服务类下的服务方法，当我们以订阅的方式激活此Actor的时候，是以被订阅的服务方法的参数来区分的，简单说，就是订阅的服务方法参数一样，那么多个客户端订阅的都是同一个MSF的服务对象实例。

这个现象，可以通过本篇的“闹钟服务”订阅过程来验证，在第一个客户端订阅闹钟服务后，启动第二个TestClient程序，也来订阅闹钟服务，注意，2个进程订阅的闹钟服务，它的闹铃时间设置为一样。订阅后，我们发现，即使第一个订阅客户端已经开始收到服务器的“闹铃消息”推送，第二个订阅客户端加入进来后，可以马上收到同样的消息推送，这说明，两个客户端订阅的是同一个MSF的服务对象，也就是同一个Actor对象。我们注意观察 MSF Host的屏幕输出，也能验证这个结果，它会提示消息发送给了2个客户端，具体过程，大家可以去仔细看看，本篇不再说明。下面是效果图：

---------------------------分界线------------------------------------------------------------------------

欢迎加入我们的QQ群讨论MSF框架的使用，群号：敏思（PWMIS） .NET 18215717，加群请注明：PDF.NET技术交流，否则可能被拒。

本文转自深蓝医生博客园博客，原文链接：http://www.cnblogs.com/bluedoctor/p/7649169.html，如需转载请自行联系原作者

“一切都是消息”--MSF（消息服务框架）之【发布-订阅】模式

一、定时推送模式

1.1，编写“时间服务”

1.2，在TestClient 项目添加订阅服务的代码：

1.3，改变推送频率

二、事件推送模式

2.1,编写闹钟服务

2.2，编写闹铃服务订阅客户端

2.3，注册MSF服务方法的参数类

2.4，运行订阅服务

三、MSF的Actor模式

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

“一切都是消息”--MSF（消息服务框架）之【发布-订阅】模式

一、定时推送模式

1.1，编写“时间服务”

1.2，在TestClient 项目添加订阅服务的代码：

1.3，改变推送频率

二、事件推送模式

2.1,编写闹钟服务

2.2，编写闹铃服务订阅客户端

2.3，注册MSF服务方法的参数类

2.4，运行订阅服务

三、MSF的Actor模式

热门文章

最新文章

相关课程

相关电子书

相关实验场景