Publisher/Subscriber 订阅-发布模式
本博后续将陆续整理这些年做的一些预研demo，及一些前沿技术的研究，与大家共研技术，共同进步。
关于发布订阅有很多种实现方式，下面主要介绍WCF中的发布订阅，主要参考书籍《Programming WCF Services》，闲话不多说进入正题。使用传统的双工回调（例子 http://www.cnblogs.com/artech/archive/2007/03/02/661969.html）实现发布订阅模式存在许多缺陷，主要问题是，它会引入发布者和订阅者之间的高度耦合。订阅者必须先知道发布者在哪里，然后才能订阅它们，任何订阅者不知道的服务都无法通知事件的订阅者，部署好的应用程序中添加新的订阅者（或者移除已经存在的订阅者）是十分困难的事情。大致相同的是发布者也只能给它知道的订阅者发送通知消息，同时发布者还需要管理订阅者列表，这些与业务服务无关，这些逻辑增加了发布者的复杂度，另外在安全方面也存在订阅者与发布者也存在耦合，而且在发布者进程宕机时，所有订阅都会丢失。
要解决上面提及的问题最常见的解决方案就是发布-订阅模式（Publish-Subscribe 【OBSERVER】），如图D-1所示。
这里将订阅者区分为临时订阅者与持久订阅者，持久订阅者可以保存到磁盘上，当事件触发时可以通知订阅者，也可以很方便的通过传递回调使用回调机制，对于持久订阅者，需要记录订阅者地址，当触发事件时，发布服务将会调用持久订阅者地址 ，然后传递事件，因持久订阅者保存了订阅者地址至数据库或磁盘，因此当发布服务宕机时提高了管理性。
以上主要介绍理论，下面进入实践阶段，首先下载ServiceModelEx(Programming WCF Services 里面书籍作者提供的简化WCF编程的动态库), https://github.com/CaseyBurns/ServiceModelEx，我们暂时不需要服务总线所以我们引入ServiceModelEx (.NET 4.0 no service bus) ，建好测试服务端(这里为了方便测试使用GUI 应用程序作为宿主)，客户端。
管理临时订阅
例子D-1使用ServiceModelEx 提供的ISubscriptionService接口管理临时订阅者
[ServiceContract]
public interface ISubscriptionService
{
[OperationContract]
void Subscribe(string eventOperation);
[OperationContract]
void Unsubscribe(string eventOperation);
}
作为通用接口它不关心回调契约,然后添加临时订阅者契约继承通用接口,并设置回调契约
[ServiceContract(CallbackContract = typeof(IMyEvents))]
public interface IMySubscriptionService : ISubscriptionService
{
}
回调契约
[ServiceContract]
public interface IMyEvents
{
[OperationContract(IsOneWay = true)]
void OnEvent1();
[OperationContract(IsOneWay = true)]
void OnEvent2(int number);
[OperationContract(IsOneWay = true)]
void OnEvent3(int number, string text);
}
实现临时订阅服务.
[ServiceBehavior(InstanceContextMode = InstanceContextMode.PerCall)]
public class MySubscriptionService : SubscriptionManager<IMyEvents>, IMySubscriptionService,IPersistentSubscriptionService
{
}
这里有几点需要注意：服务类型必须是会话服务(InstanceContextMode = InstanceContextMode.PerCall)，会话服务才能够使用回调，另外ServiceModelEx 中的类 SubscriptionManager<T> 已经实现了通用接口所定义的添加订阅者与取消订阅接口,所以这里不需要我们再写任何代码。IPersistentSubscriptionService 作为持久订阅者接口，SubscriptionManager<T> 也实现了该接口，接下来会讲到。
配置文件配置发布订阅者服务
<system.serviceModel>
<serviceHostingEnvironment multipleSiteBindingsEnabled="true" />
<bindings>
<netTcpBinding>
<binding name="NetTcpBinding_IService1" receiveTimeout="00:25:00"
maxBufferSize="2147483647" maxReceivedMessageSize="2147483647" transactionFlow="true">
<reliableSession inactivityTimeout="00:25:00" enabled="true" />
<security mode="None" />
</binding>
</netTcpBinding>
</bindings>
<services>
<service behaviorConfiguration="MyBehavior" name="Service.sub.MySubscriptionService">
<host>
<baseAddresses>
<add baseAddress="net.tcp://localhost:8022/"/>
<add baseAddress="http://localhost:8023/"/>
</baseAddresses>
</host>
<endpoint name="Sub" address="Sub" binding="netTcpBinding" bindingConfiguration="NetTcpBinding_IService1"
contract="Service.sub.IMySubscriptionService" />
<endpoint name="PersistentSub" address="PersistentSub" binding="netTcpBinding" bindingConfiguration="NetTcpBinding_IService1"
contract="ServiceModelEx.IPersistentSubscriptionService" />
</service>
<service behaviorConfiguration="MyBehavior" name="Service.pub.MyPublishService">
<host>
<baseAddresses>
<add baseAddress="net.tcp://localhost:8022/MyPub/"/>
<add baseAddress="http://localhost:8023/MyPub/"/>
</baseAddresses>
</host>
<endpoint name="PubMyEvents" address="PubMyEvents" binding="netTcpBinding" bindingConfiguration="NetTcpBinding_IService1"
contract="Service.sub.IMyEvents" />
</service>
</services>
<behaviors>
<serviceBehaviors>
<behavior name="MyBehavior">
<serviceMetadata httpGetEnabled="true"/>
<serviceThrottling maxConcurrentCalls="1000" maxConcurrentSessions="10000" />
<serviceDebug includeExceptionDetailInFaults="true" />
</behavior>
</serviceBehaviors>
</behaviors>
</system.serviceModel>
其中Service.pub.MyPublishService 服务为发布者服务配置 接下来会讲到。
这样临时订阅者就实现了，接下来看持久订阅者.持久订阅者的通用接口使用ServiceModelEx中定义的IPersistentSubscriptionService
[ServiceContract]
public interface IPersistentSubscriptionService
{
[OperationContract(Name = "SubscribePersistent")]
[TransactionFlow(TransactionFlowOption.Allowed)]
void Subscribe(string address,string eventsContract,string eventOperation);
[OperationContract(Name = "UnSubscribePersistent")]
[TransactionFlow(TransactionFlowOption.Allowed)]
void Unsubscribe(string address,string eventsContract,string eventOperation);
[OperationContract]
[TransactionFlow(TransactionFlowOption.Allowed)]
PersistentSubscription[] GetAllSubscribers();
[OperationContract]
[TransactionFlow(TransactionFlowOption.Allowed)]
PersistentSubscription[] GetSubscribersToContract(string eventsContract);
[OperationContract]
[TransactionFlow(TransactionFlowOption.Allowed)]
string[] GetSubscribersToContractEventType(string eventsContract,string eventOperation);
[OperationContract]
[TransactionFlow(TransactionFlowOption.Allowed)]
PersistentSubscription[] GetAllSubscribersFromAddress(string address);
}
这里我添加了对[OperationContract(Name = "SubscribePersistent")] 将添加订阅方法进行重命名，以区别临时订阅接口的Subscribe方法.持久订阅不需要回调函数，接下来实现持久订阅同样简单，上面已经贴过代码，ServiceModelEx中的SubscriptionManager<T>同样已经实现了IPersistentSubscriptionService接口，这样临时订阅与持久订阅完成，接下来看发布服务。
发布服务应该支持与订阅服务一样的事件契约，这是订阅服务与发布服务唯一的连接点，使用IMyEvents 作为例子，另外ServiceModelEx提供了用于简化发布服务的帮助类PublishService<T>
[ServiceBehavior(InstanceContextMode = InstanceContextMode.PerCall)]
public class MyPublishService : PublishService<IMyEvents>, IMyEvents
{
public void OnEvent1()
{
FireEvent();
}
public void OnEvent2(int number)
{
FireEvent(number);
}
public void OnEvent3(int number, string text)
{
FireEvent(number, text);
}
}
其中FireEvent()被用作激发所有订阅者的事件，无论是临时还是持久订阅者，帮助类PublishService<T>已经做了实现，接下来配置发布服务
<service behaviorConfiguration="MyBehavior" name="Service.pub.MyPublishService">
<host>
<baseAddresses>
<add baseAddress="net.tcp://localhost:8022/MyPub/"/>
<add baseAddress="http://localhost:8023/MyPub/"/>
</baseAddresses>
</host>
<endpoint name="PubMyEvents" address="PubMyEvents" binding="netTcpBinding" bindingConfiguration="NetTcpBinding_IService1"
contract="Service.sub.IMyEvents" />
</service>
这样发布服务完成，使用Gui应用程序作为宿主，可以使用ServiceModelEx 中ServiceHost<T> 作为发布的帮助类 。
public partial class Form1 : Form
{
public Form1()
{
InitializeComponent();
}
ServiceHost<MyPublishService> hostPub = new ServiceHost<MyPublishService>();
ServiceHost<MySubscriptionService> host = new ServiceHost<MySubscriptionService>();
private void Form1_Load(object sender, EventArgs e)
{
try
{
host.EnableMetadataExchange();
host.Open();
hostPub.EnableMetadataExchange();
hostPub.Open();
}
catch (Exception ex)
{
throw;
}
}
private void Form1_FormClosed(object sender, FormClosedEventArgs e)
{
try
{
host.Close();
}
catch (Exception)
{
try
{
host.Abort();
}
catch (Exception)
{
}
}
try
{
hostPub.Close();
}
catch (Exception)
{
try
{
hostPub.Abort();
}
catch (Exception)
{
}
}
}
}
其中 host.EnableMetadataExchange(); 能够帮助发布元数据，不需要再到配置中进行配置,服务配置好后接下来看客户端使用,
客户端可以直接添加服务引用生成服务代理，但是一般本人喜欢使用SvcUtil工具生成代理，或者干脆直接使用通道进行服务调用，后者更为我所喜爱，因为这样代码阅读行更强，更简练。例子中偷了下懒，直接添加服务引用，然后用通道调用服务，这样剩了点复制配置或者接口的功夫，所以看到例子不要感到奇怪，全因懒造成的，废话不多说，接下来看临时订阅客户端调用
DuplexChannelFactory<IMySubscriptionService, IMySubscriptionServiceCallback> channelFactory = null;
IMySubscriptionService proxy = null;
private void btnSub_Click(object sender, EventArgs e)
{
MyEventsCallback callBack = new MyEventsCallback();
callBack.OnResultEvent += CallBack_OnResultEvent;
InstanceContext<IMySubscriptionServiceCallback> instanceContext = new InstanceContext<IMySubscriptionServiceCallback>(callBack);
channelFactory = new DuplexChannelFactory<IMySubscriptionService, IMySubscriptionServiceCallback>(instanceContext, "Sub");
proxy = channelFactory.CreateChannel();
proxy.Subscribe(null);
}
这里使用ServiceModelEx 中提供的DuplexChannelFactory<T,C> 类型安全的双向通道类创建代理，MyEventsCallback 实现回调接口，具体实现如下：
internal class MyEventsCallback : IMySubscriptionServiceCallback
{
SynchronizationContext sc = SynchronizationContext.Current;
public event EventHandler<EventsCallbackArgs> OnResultEvent;
public void OnEvent1()
{
sc.Post(result =>
{
EventsCallbackArgs e = new EventsCallbackArgs() { Msg = string.Concat("OnEvent1", System.Environment.NewLine) };
e.Raise(this, ref OnResultEvent);
}, null);
}
public void OnEvent2(int number)
{
sc.Post(result =>
{
EventsCallbackArgs e = new EventsCallbackArgs() { Msg = string.Concat("OnEvent2:", number, System.Environment.NewLine) };
e.Raise(this, ref OnResultEvent);
}, null);
}
public void OnEvent3(int number, string text)
{
sc.Post(result =>
{
EventsCallbackArgs e = new EventsCallbackArgs() { Msg = string.Concat("OnEvent3:", number, "text:", text + System.Environment.NewLine) };
e.Raise(this, ref OnResultEvent);
}, null);
}
}
public static class EventArgExtensions
{
public static void Raise<TEventArgs>(this TEventArgs e, Object sender, ref EventHandler<TEventArgs> eventDelegate) where TEventArgs : EventArgs
{
EventHandler<TEventArgs> temp = Interlocked.CompareExchange(ref eventDelegate, null, null);
if (temp != null) temp(sender, e);
}
}
SynchronizationContext 上下文提供post方法调用gui线程更新ｕｉ，e.Raise 使用扩展以线程安全方式调用事件，客户端调用订阅者就完成了，别忘了关闭代理，接下来看客户端调用发布者
客户端调用发布服务：
public partial class PubMessageForm : Form
{
public PubMessageForm()
{
InitializeComponent();
}
ChannelFactory<IMyEvents> channelFactory = null;
IMyEvents proxy = null;
private void btnStartPub_Click(object sender, EventArgs e)
{
channelFactory = new ChannelFactory<IMyEvents>("PubMyEvents");
proxy = channelFactory.CreateChannel();
}
private void PubMessageForm_FormClosed(object sender, FormClosedEventArgs e)
{
try
{
using (proxy as IDisposable)
{
}
channelFactory.Close();
}
catch
{
channelFactory.Abort();
}
}
private void btnPub_Click(object sender, EventArgs e)
{
proxy.OnEvent1();
}
private void btnPub2_Click(object sender, EventArgs e)
{
proxy.OnEvent2(2);
}
private void btnPub3_Click(object sender, EventArgs e)
{
proxy.OnEvent3(3, txtPubMessage.Text);
}
private void PubMessageForm_Load(object sender, EventArgs e)
{
}
}
使用ChannelFactory<T> 通道调用发布服务
这样WCF发布订阅服务就完成了，另外如果发布服务或订阅服务不需要同步绑定，可以考虑使用msmq ，这样发布-订阅模式兼具松耦合和无连接系统的优势。
需要注意的是队列化发布-订阅服务不支持临时订阅，需要使用持久订阅，具体实现在此不多讲，另外还可以结合服务发现实现另外一种模式的发布订阅模式，具体可以参考书籍《Programming WCF Services》。
Demo下载 http://files.cnblogs.com/files/skystar/Demo.7z
书中自有黄金屋，书中自有颜如玉