回调与并发: 通过实例剖析WCF基于ConcurrencyMode.Reentrant模式下的并发控制机制

对于正常的服务调用,从客户端发送到服务端的请求消息最终会被WCF服务运行时分发到相应的封装了服务实例的InstanceContext上。而在回调场景中,我们同样将回调对象封装到InstanceContext对象,并将其封送到客户端。当服务操作过程中执行回调操作的时候,回调消息最终也是分发到位于客户端封装回调对象的InstanceContext。从消息分发与并发处理的机制来看,这两种请求并没有本质的不同。接下来,我们通过《实践重于理论》中的实例,综合分析WCF对并发服务调用和并发回调的处理机制。

一、将实例改成支持回调的形式

为此,我们需要对我们上面给出的监控程序进行相应的修改。首先需要修改的是服务契约ICalculator。服务契约ICalculator的Add操作接受传入的操作数并以返回值得形式返回到客户端。现在我们通过回调的形式来重写计算服务:将Add的返回类型改称void,计算结果通过执行回调操作的形式在客户端显示。

   1: [ServiceContract(Namespace="http://www.artech.com/",CallbackContract =typeof(ICalculatorCallback))]
   2: public interface ICalculator
   3: {
   4:     [OperationContract]
   5:     void Add(double x, double y);
   6: }

作为回调契约的ICalculatorCallback接口定义如下,计算结果传入ShowResult方法显示出来。在一般情况下,我们会将Add和ShowResult和操作定义在单向(One-way),但是这里我并没有这么做,所以无论是服务操作Add还是回调操作ShowResult均采用请求/回复消息交换模式。

   1: using System.ServiceModel;
   2: namespace Artech.ConcurrentServiceInvocation.Service.Interface
   3: {
   4:     [ServiceContract(Namespace = "http://www.artech.com/")]
   5:     public interface ICalculatorCallback
   6:     {
   7:         [OperationContract]
   8:         void ShowResult(double result);
   9:     }
  10: }

在本例中我们的CalculatorService采用单例实例上下文模式(InstanceContextMode.Single)。为了能够执行回调,将并发模式设置成ConcurrencyMode.Reentrant。在Add操作中,我们可以将整个执行过程分成三个阶段:PreCallback、Callback和PostCallback,而且PreCallback和PostCallback执行时间为5秒。在开始和结束执行Add操作,以及开始与结束回调的时候都是通过EventMonitor发送相应的事件通知。修改后的CalculatorService如下面的代码所示。

   1: [ServiceBehavior(UseSynchronizationContext = false,InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Reentrant)]
   2: public class CalculatorService : ICalculator
   3: {
   4:     public void Add(double x, double y)
   5:     {
   6:         //PreCallback
   7:         EventMonitor.Send(EventType.StartExecute);
   8:         Thread.Sleep(5000);
   9:         double result = x + y;
  10:  
  11:         //Callback
  12:         EventMonitor.Send(EventType.StartCallback);
  13:         int clientId = OperationContext.Current.IncomingMessageHeaders.GetHeader<int>(EventMonitor.CientIdHeaderLocalName, EventMonitor.CientIdHeaderNamespace);
  14:         MessageHeader<int> messageHeader = new MessageHeader<int>(clientId);
  15:         OperationContext.Current.OutgoingMessageHeaders.Add(messageHeader.GetUntypedHeader(EventMonitor.CientIdHeaderLocalName, EventMonitor.CientIdHeaderNamespace));
  16:         OperationContext.Current.GetCallbackChannel<ICalculatorCallback>().ShowResult(result);
  17:         EventMonitor.Send(EventType.EndCallback);
  18:  
  19:         //PostCallback
  20:         Thread.Sleep(5000);
  21:         EventMonitor.Send(EventType.EndExecute);
  22:     }
  23: }

对于服务寄宿程序我们不需要做任何修改,但是我们需要采用支持双向通信的绑定类型以实现对回调的支持,在这里我们采用的是NetTcpBinding。为了降低安全协商(Negotiation)代码对时延,我特意将绑定的安全模式设置成None。下面是更新后的服务端配置,客户端需要进行相应的修改。

   1: <?xml version="1.0" encoding="utf-8" ?>
   2: <configuration>
   3:     <system.serviceModel>
   4:       <bindings>
   5:         <netTcpBinding>
   6:           <binding name="nonSecureBinding">
   7:             <security mode="None" />
   8:           </binding>
   9:         </netTcpBinding>
  10:       </bindings>
  11:         <services>
  12:             <service name="Artech.ConcurrentServiceInvocation.Service.CalculatorService">
  13:                 <endpoint address="net.tcp://127.0.0.1:3721/calculatorservice" binding="netTcpBinding"
  14:                     bindingConfiguration="nonSecureBinding" contract="Artech.ConcurrentServiceInvocation.Service.Interface.ICalculator" />
  15:             </service>
  16:         </services>
  17:     </system.serviceModel>
  18: </configuration>

由于回调操组在客户端执行,所以客户端首先需要的就是实现回调契约接口创建回调类型。实现回调契约接口的ICalculatorCallback定义在CalculatorCallbackService类型中。由于在本例中我们需要的仅仅监控回调操作执行的时间,并不是真的需要显示出运算的最终结果。所以我们仅仅是通过挂起当前线程模拟一个耗时的回调操作(10秒),在回调操作开始和结束执行的时候通过EventMonitor发送相应的事件通知。

   1: using System.ServiceModel;
   2: using System.Threading;
   3: using Artech.ConcurrentServiceInvocation.Service.Interface;
   4: namespace Artech.ConcurrentServiceInvocation.Client
   5: {
   6:     public class CalculatorCallbackService : ICalculatorCallback
   7:     {
   8:         public void ShowResult(double result)
   9:         {
  10:             EventMonitor.Send(EventType.StartExecuteCallback);
  11:             Thread.Sleep(10000);
  12:             EventMonitor.Send(EventType.EndExecuteCallback);
  13:         }
  14:     }
  15: }

最后一个步骤是对客户端按照回调的方式进行相应的修改。首先我们创建CalculatorCallbackService对象,并以此创建一个InstanceContext作为回调实例上下文。然后通过该InstanceContext创建DuplexChannelFactory<TChannel>。最后通过ThreadPool并发地执行2次服务代理的创建和服务调用的操作,客户端ID作为消息报头被传送到服务端。

   1: public partial class MonitorForm : Form
   2: {
   3:     private SynchronizationContext _syncContext;
   4:     private DuplexChannelFactory<ICalculator> _channelFactory;
   5:     private InstanceContext _callbackInstance;
   6: private int _clientId = 0;
   7:  
   8:     //其他成员
   9:     private void MonitorForm_Load(object sender, EventArgs e)
  10:     {
  11:         string header = string.Format("{0, -13}{1, -22}{2}", "Client", "Time", "Event");
  12:         this.listBoxExecutionProgress.Items.Add(header);
  13:         _syncContext = SynchronizationContext.Current;
  14:         _callbackInstance = new InstanceContext(new CalculatorCallbackService());
  15:         _channelFactory = new  DuplexChannelFactory<ICalculator>(_callbackInstance,"calculatorservice");
  16:  
  17:         EventMonitor.MonitoringNotificationSended += ReceiveMonitoringNotification;
  18:         this.Disposed += delegate
  19:         {
  20:             EventMonitor.MonitoringNotificationSended -= ReceiveMonitoringNotification;
  21:             _channelFactory.Close();
  22:         };
  23:  
  24:         for (int i = 0; i < 2; i++)
  25:         {
  26:             ThreadPool.QueueUserWorkItem(state =>
  27:             {
  28:                 int clientId = Interlocked.Increment(ref _clientId);
  29:                 EventMonitor.Send(clientId, EventType.StartCall);
  30:                 ICalculator proxy = _channelFactory.CreateChannel();
  31:                 using (OperationContextScope contextScope = new OperationContextScope(proxy as IContextChannel))
  32:                 {
  33:                     MessageHeader<int> messageHeader = new MessageHeader<int>(clientId);
  34:                     OperationContext.Current.OutgoingMessageHeaders.Add(messageHeader.GetUntypedHeader(EventMonitor.CientIdHeaderLocalName, EventMonitor.CientIdHeaderNamespace));
  35:                     proxy.Add(1, 2);
  36:                 }
  37:                 EventMonitor.Send(clientId, EventType.EndCall);
  38:             }, null);
  39:         }
  40:     } 
  41: }

二、从并发控制机制分析得到的输出结果

现在重新运行我们更新后的监控程序,你将会得到如图1所示的输出结果。如果你仔细分析服务端和客户端输出的结果你将会看到Add操作的整个执行时间有一段是重合的,也就是说整个服务操作存在并发执行的情况。但是单看PreCallback和PostCallback,则不存在并发执行的情况。从客户端的角度来看,回调操作也不存在并发执行的情况

图1 Reentrant(Service) + Single(Callback)监控结果

可能上面的输出结果还不是很直观,现在我们通过时间轴的形式来描述通过输出结果表现出的执行情况。我们忽略掉客户端和服务通信以及WCF消息分发导致的时延,两次服务调用在执行的情况如图2所示。假设服务端在0s接收到两个并发的调用请求,一个请求被分发给InstanceContext,另一个则被放到等待队列。到5s的时候,第一个请求完成PreCallback的操作后进行回调,此时InstanceContext被释放出来,使得它可以用于处理等待着的第二个请求。到10s的时候,第二个请求完成了PreCallback操作准备进行回调,但是封装回调实例的InstanceContext正在处理第一个回调请求,所示自己在一个等待,直到20s时第一个回调请求处理完毕。

图2 Reentrant(Service) + Single(Callback)监控结果时间轴描述

上面我们模拟的时单例实例上下文情况下,服务和回调分别采用Concurrency.Reentrant和Concurrency.Single的情况。实例演示的结果充分证明在《并发中的同步--WCF并发体系的同步机制实现》中关于针对InstanceContext加锁的同步机制的分析。进一步地,如果按照我们的分析,如果我们同时将服务和回调采用的并发模式均换成Concurrency.Multiple,那么无论是作用于服务实例上下文的PreCallback和PostCallback操作,还是作用于回调实例上下文的Callback都可以并发地执行。为此,我们只需要对分别应用于CalculatorService和CalculatorCallbackService的ServiceBehaviorAttribute和CallbackBehaviorAttribute的两个特性稍加修改,将ConcurrencyMode属性设置成ConcurrencyMode.Multiple即可。相应的改动如下面的代码所示:

   1: [ServiceBehavior(UseSynchronizationContext = false,InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Multiple)]
   2: public class CalculatorService : ICalculator
   3: {
   4:    //省略成员
   5: }
   6:  
   7: [CallbackBehavior(ConcurrencyMode = ConcurrencyMode.Multiple)]
   8: public class CalculatorCallbackService : ICalculatorCallback
   9: {
  10: //省略成员
  11: }

再次运行我们的监控程序,得到的如图3所示的输出,可以看出这正是我们希望的结果,无论作用于那个InstanceContext的操作都是并发执行的

图3 Multiple(Service) + Multiple(Callback)监控结果

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏圣杰的专栏

事件总线知多少(2)

源码路径:Github-EventBus 事件总线知多少(1) 事件总线知多少(2) 1.引言 之前的一篇文章事件总线知多少(1),介绍了什么是事件总线...

2367
来自专栏运维一切

laravel自定义错误页面 原

app\Exceptions\handler.php 在render的时候就携带了这个异常

1213
来自专栏坚毅的PHP

jersey处理支付宝异步回调通知的问题:java.lang.IllegalArgumentException: Error parsing media type 'application/x-www

tcpflow以流为单位分析请求内容,非常适合服务器端接口类服务查问题 这次遇到的问题跟支付宝支付后的回调post结果有关 淘宝的代码例子: publi...

6105
来自专栏恰童鞋骚年

设计模式的征途—19.命令(Command)模式

在生活中,我们装修新房的最后几道工序之一是安装插座和开关,通过开关可以控制一些电器的打开和关闭,例如电灯或换气扇。在购买开关时,用户并不知道它将来到底用于控制什...

682
来自专栏一个爱瞎折腾的程序猿

在asp.net core2.1中添加中间件以扩展Swashbuckle.AspNetCore3.0支持简单的文档访问权限控制

在此之前的接口项目中,若使用了 Swashbuckle.AspNetCore,都是控制其只在开发环境使用,不会就这样将其发布到生产环境(安全第一) 。 那么,...

1741
来自专栏大内老A

ASP.NET MVC集成EntLib实现“自动化”异常处理[实例篇]

个人觉得异常处理对于程序员来说是最为熟悉的同时也是最难掌握的。说它熟悉,因为仅仅就是try/catch/finally而已。说它难以掌握,则是因为很多开发人员却...

21810
来自专栏lzj_learn_note

Volley源码分析学习

2)根据SDK版本来创建HttpStack的实现,如果是2.3以上的,则使用基于HttpUrlConnection实现的HurlStack,反之,则利用Http...

1176
来自专栏大内老A

[WCF REST] 通过ASP.NET Output Caching实现声明式缓存

ASP.NET的输出缓存(Output Caching)机制允许我们针对整个Web页面或者页面的某个部分(主要针对用户控件)最终呈现的HTML进行缓存。对于后续...

1916
来自专栏木木玲

Netty 那些事儿 ——— 心跳机制

7989
来自专栏博客园

.NET Core 使用RabbitMQ

  RabbitMQ是一个开源的,基于AMQP(Advanced Message Queuing Protocol)协议的完整,可复用的企业级消息队列(Mess...

1513

扫码关注云+社区

领取腾讯云代金券