當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

WCF后续之旅(11): 关于并发、回调的线程关联性（Thread Affinity）

發(fā)布時(shí)間：2023/12/9 编程问答 40 豆豆

生活随笔收集整理的這篇文章主要介紹了 WCF后续之旅(11): 关于并发、回调的线程关联性（Thread Affinity）小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

對(duì)于一般的多線程操作，比如異步地進(jìn)行基于文件系統(tǒng)的IO操作；異步地調(diào)用Web Service；或者是異步地進(jìn)行數(shù)據(jù)庫訪問等等，是和具體的線程無關(guān)的。也就是說，對(duì)于這些操作，任意創(chuàng)建一個(gè)新的線程來執(zhí)行都是等效的。但是有些情況下，有些操作卻只能在固定的線程下執(zhí)行。比如，在GUI應(yīng)用下，對(duì)控件的訪問就需要在創(chuàng)建該控件的線程下執(zhí)行；或者我們?cè)谀硞€(gè)固定的線程中通過TLS（Thread Local Storage）設(shè)置了一些Context信息，供具體的操作使用，我們把操作和某個(gè)固定的線程的依賴稱為線程關(guān)聯(lián)性（Thread Affinity）。在這種情況下，我們的異步操作就需要被Marshal到固定的線程執(zhí)行。在WCF并發(fā)或者Callback的情況下也具有這樣的基于線程關(guān)聯(lián)性的問題。

一、從基于Windows Application客戶端的WCF回調(diào)失敗談起

在"我的WCF之旅"系列文章中,有一篇（WinForm Application中調(diào)用Duplex Service出現(xiàn)TimeoutException的原因和解決方案）專門介紹在一個(gè)Windows Application客戶端應(yīng)用, 通過WCF 的Duplex通信方式進(jìn)行回調(diào)失敗的文章.我們今天以此作為出發(fā)點(diǎn)介紹WCF在Thread Affinity下的表現(xiàn)和解決方案.

我們來創(chuàng)建一個(gè)WCF的應(yīng)用來模擬該場景: 客戶端是一個(gè)基于Windows Form應(yīng)用, 完成一個(gè)計(jì)算器的功能, 用戶輸入操作數(shù),點(diǎn)擊"計(jì)算"按鈕, 后臺(tái)通過調(diào)用WCF service, 并傳遞一個(gè)用于顯示計(jì)算結(jié)果的Callback對(duì)象; service進(jìn)行相應(yīng)的計(jì)算得到最后的運(yùn)算結(jié)果,調(diào)用該Callback對(duì)象將運(yùn)算結(jié)果顯示到客戶端界面.這是我們的WCF四層結(jié)構(gòu)：

1、Contract：ICalculate & ICalculateCallback

1: namespace Artech.ThreadAffinity.Contracts 2: { 3: [ServiceContract(CallbackContract = typeof(ICalculateCallback))] 4: public interface ICalculate 5: { 6: [OperationContract] 7: void Add(double op1, double op2); 8: } 9: }

這是Service Contract，下面是Callback Contract，用于顯示運(yùn)算結(jié)果：

1: namespace Artech.ThreadAffinity.Contracts 2: { 3: public interface ICalculateCallback 4: { 5: [OperationContract] 6: void DisplayResult(double result); 7: } 8: }

2、Service：CalculateService

1: namespace Artech.ThreadAffinity.Services 2: { 3: [ServiceBehavior(ConcurrencyMode = ConcurrencyMode.Reentrant)] 4: public class CalculateService:ICalculate 5: { 6: public static ListBox DisplayPanel 7: { get; set; } 8:? 9: #region ICalculate Members 10:? 11: public void Add(double op1, double op2) 12: { 13: double result = op1 + op2; 14: ICalculateCallback callback = OperationContext.Current.GetCallbackChannel<ICalculateCallback>(); 15:? 16: DisplayPanel.Items.Add(string.Format("{0} + {1} = {2}", op1, op2, result)); 17:? 18: callback.DisplayResult(result); 19: } 20:? 21: #endregion 22: } 23: }

由于需要進(jìn)行callback, 我們把ConcurrencyMode 設(shè)為Reentrant。當(dāng)?shù)玫竭\(yùn)算的結(jié)果后，通過OperationContext.Current.GetCallbackChannel得到callback對(duì)象，并調(diào)用之。還有一點(diǎn)需要提的是，該service是通過一個(gè)Windows Form application進(jìn)行host的。并且有一個(gè)ListBox列出所有service執(zhí)行的結(jié)果，就像這樣：

3、Hosting

Hosting的代碼寫在Form的Load事件中：

1: private void HostForm_Load(object sender, EventArgs e) 2: { 3: this._serviceHost = new ServiceHost(typeof(CalculateService)); 4: CalculateService.DisplayPanel = this.listBoxResult; 5: CalculateService.SynchronizationContext = SynchronizationContext.Current; 6: this._serviceHost.Opened += delegate 7: { 8: this.Text = "The calculate service has been started up!"; 9: }; 10:? 11: this._serviceHost.Open(); 12: }

我們注意到了CalculateService使用到的用于顯示所有預(yù)算結(jié)果的ListBox就是在這了通過static property傳遞的。

這么配置文件

1: <configuration> 2: <system.serviceModel> 3: <services> 4: <service name="Artech.ThreadAffinity.Services.CalculateService"> 5: <endpoint binding="netTcpBinding" bindingConfiguration="" contract="Artech.ThreadAffinity.Contracts.ICalculate" /> 6: <host> 7: <baseAddresses> 8: <add baseAddress="net.tcp://127.0.0.1:8888/calculateservice" /> 9: </baseAddresses> 10: </host> 11: </service> 12: </services> 13: </system.serviceModel> 14: </configuration>

4、Client

Client的界面很簡單：輸入兩個(gè)操作數(shù)，點(diǎn)擊“=”按鈕，將運(yùn)算結(jié)果顯示出來。

先來看看client端對(duì)callback contract的實(shí)現(xiàn)：

1: namespace Clients 2: { 3: public class CalculateCallback : ICalculateCallback 4: { 5: public static TextBox ResultPanel; 6:? 7: #region ICalculateCallback Members 8:? 9: public void DisplayResult(double result) 10: { 11: ResultPanel.Text = result.ToString(); 12: } 13:? 14: #endregion 15: } 16: }

這是配置：

1: <configuration> 2: <system.serviceModel> 3: <client> 4: <endpoint address="net.tcp://127.0.0.1:8888/calculateservice" 5: binding="netTcpBinding" bindingConfiguration="" contract="Artech.ThreadAffinity.Contracts.ICalculate" 6: name="calculateservice" /> 7: </client> 8: </system.serviceModel> 9: </configuration> 然后是我們“=”按鈕的單擊事件對(duì)運(yùn)算的實(shí)現(xiàn)： 1: private void buttonCalculate_Click(object sender, EventArgs e) 2: { 3: CalculateCallback.ResultPanel = this.textBoxResult; 4: DuplexChannelFactory<ICalculate> channelFactory = new DuplexChannelFactory<ICalculate>(new CalculateCallback(), "calculateservice"); 5: ICalculate calculator = channelFactory.CreateChannel(); 6: calculator.Add(double.Parse(this.textBoxOp1.Text), double.Parse(this.textBoxOp2.Text)); 7: }

CalculateCallback 用于顯示運(yùn)算結(jié)果的TextBox通過statis property實(shí)現(xiàn)傳遞。這個(gè)實(shí)現(xiàn)很簡單，貌似沒有什么問題，但是我們運(yùn)行程序，在客戶端就會(huì)拋出這樣的exception。可以看出是一個(gè)TimeoutException。

二、是什么導(dǎo)致TimeoutException？

我們現(xiàn)在來分析是什么導(dǎo)致了TimeoutException的拋出。原因很簡單：由于我們對(duì)service的調(diào)用的是在UI 線程調(diào)用的，所以在開始調(diào)用到最終得到結(jié)果，這個(gè)UI Thread會(huì)被鎖住；但是當(dāng)service進(jìn)行了相應(yīng)的運(yùn)算的到運(yùn)算的結(jié)果后，需要調(diào)用callback對(duì)象對(duì)client進(jìn)行回調(diào)，默認(rèn)的情況下，Callback的執(zhí)行是在UI線程執(zhí)行的。當(dāng)Callback試圖執(zhí)行的時(shí)候，發(fā)現(xiàn)UI 線程被鎖，只能等待。這樣形成一個(gè)死鎖，UI線程需要等待CalculateService執(zhí)行返回后才能解鎖，而CalculateService需要Callback執(zhí)行完成；而Callback需要等到UI線程解鎖才能執(zhí)行。

基于上門的原因，我們有兩種解決方案：

CalculateService不必等到Callback執(zhí)行完成就返回，我們可以通過異步調(diào)用Callback。或者讓Client異步方式調(diào)用CalculateService，以便及時(shí)釋放UI線程，我們可以通過One-way的方式來進(jìn)行service的調(diào)用。
讓Callback的執(zhí)行不必綁定到UI線程

三、解決方案一：通過異步調(diào)用或者One-way回調(diào)

為了簡單起見，我們通過ThreadPool實(shí)現(xiàn)了異步回調(diào)：

1: public void Add(double op1, double op2) 2: { 3: double result = op1 + op2; 4: ICalculateCallback callback = OperationContext.Current.GetCallbackChannel<ICalculateCallback>(); 5:? 6: ThreadPool.QueueUserWorkItem(delegate{ callback.DisplayResult(result); }, null); 7: }

這是一種方案，另一種是將Add操作設(shè)成One-way的：

這兩種方案都可以解決問題。

四、方案二、通過解除Callback操作和UI線程的關(guān)聯(lián)性

現(xiàn)在我們才進(jìn)入我們今天討論的主題：WCF并發(fā)操作的線程關(guān)聯(lián)性問題。在這之前，我們需要了解一個(gè)重要的對(duì)象：SynchonizationContext（System.Threading.SynchronizationContext）。SynchonizationContext就是為了解決這種線程關(guān)聯(lián)性問題而設(shè)計(jì)的。SynchonizationContext提供了兩個(gè)主要的API將操作和對(duì)應(yīng)的Thread關(guān)聯(lián)：Post和Send。

1: public virtual void Post(SendOrPostCallback d, object state) 2: public virtual void Send(SendOrPostCallback d, object state)

Send和Post分別以同步和異步的方式將以Delegate表示的具體的操作和SynchonizationContext對(duì)象對(duì)應(yīng)的Thread關(guān)聯(lián)，而SendOrPostCallback delegate對(duì)象代表你需要的線程關(guān)聯(lián)操作，state代表傳入delegate的參數(shù)：

public delegate void SendOrPostCallback(object state);

對(duì)于某些具有線程關(guān)聯(lián)的應(yīng)用，比如Windows Form application，在程序啟動(dòng)的時(shí)候，會(huì)設(shè)置當(dāng)前的SynchonizationContext對(duì)象（Windows Form application使用的是繼承了SynchonizationContext的WindowsFormsSynchronizationContext ：System.Windows.Forms.WindowsFormsSynchronizationContext）。當(dāng)前SynchonizationContext被成功初始化后，你就可以通過SynchonizationContext的靜態(tài)屬性Current得到它。在你自己的應(yīng)用中，如何有需要，你也可以自定義SynchonizationContext，并通過靜態(tài)方法SetSynchronizationContext將其設(shè)置為current SynchronizationContext。

對(duì)應(yīng)WCF來說，無論是host一個(gè)service，還是在調(diào)用service時(shí)制定callback，在默認(rèn)的情況下，service和callback的操作將自動(dòng)和當(dāng)前的SynchonizationContext進(jìn)行關(guān)聯(lián)（如何有的話）。也就是說，如過我們的service被host到Windows Form application下，那么service的操作將在UI 線程下執(zhí)行；同理，如何我們?cè)谝粋€(gè)Windows Forms UI線程下調(diào)用duplex service并制定callback，那么callback的最終執(zhí)行將在UI線程。

關(guān)于WCF對(duì)線程關(guān)聯(lián)性的控制，可以通過ServiceBehavior或者CallbackBehavior的UseSynchronizationContext屬性進(jìn)行設(shè)定，該屬性默認(rèn)為true，這正式WCF默認(rèn)具有線程關(guān)聯(lián)性的原因。

現(xiàn)在我們來實(shí)現(xiàn)我們的第二套方案：讓Callback的執(zhí)行不必綁定到UI線程。為此我們只需要加上如何的CallbackBehavior attribute就可以了。

1: namespace Artech.ThreadAffinity.Clients 2: { 3: [CallbackBehavior(UseSynchronizationContext = false)] 4: public class CalculateCallback : ICalculateCallback 5: { 6: public static TextBox ResultPanel; 7:? 8: #region ICalculateCallback Members 9:? 10: public void DisplayResult(double result) 11: { 12: ResultPanel.Text = result.ToString(); 13:? 14: } 15:? 16: #endregion 17: } 18: } 19:?

但是現(xiàn)在我們運(yùn)行我們的程序，將會(huì)出現(xiàn)如下的InvalidOperation異常：

原因很簡單，由于我們將callbaclk的UseSynchronizationContext 設(shè)置成false，那么callback的操作將不會(huì)再UI線程下執(zhí)行。但是我們需要運(yùn)算的結(jié)果輸入到UI的TextBox上，對(duì)UI上控件的操作需要在UI線程上執(zhí)行，顯然會(huì)拋出異常了。

為了我們引入SynchonizationContext到CalculateCallback中：將SynchonizationContext定義成一個(gè)static屬性，通過Post方法異步地實(shí)現(xiàn)對(duì)運(yùn)算結(jié)果的顯示。

SynchonizationContext在調(diào)用service的時(shí)候指定：

1: private void buttonCalculate_Click(object sender, EventArgs e) 2: { 3: CalculateCallback.ResultPanel = this.textBoxResult; 4: CalculateCallback.SynchronizationContext = SynchronizationContext.Current; 5:? 6: DuplexChannelFactory<ICalculate> channelFactory = new DuplexChannelFactory<ICalculate>(new CalculateCallback(), "calculateservice"); 7: ICalculate calculator = channelFactory.CreateChannel(); 8: calculator.Add(double.Parse(this.textBoxOp1.Text), double.Parse(this.textBoxOp2.Text)); 9: }

現(xiàn)在我們程序能夠正常運(yùn)行了。

五、另一種可選方案：通過ISynchronizeInvoke的Invoke/BeginInvoke

熟悉Windows Form編程的讀者應(yīng)該都知道，WinForm空間的基類Control（System.Windows.Forms.Control）都實(shí)現(xiàn)了System.ComponentModel.ISynchronizeInvoke接口，而Control對(duì)ISynchronizeInvoke的實(shí)現(xiàn)就是為了解決Control的操作必須在創(chuàng)建Control線程的問題，ISynchronizeInvoke定義Invoke和BeginInvoke方法方面我們以同步或者異步的方式操作Control：

1: public interface ISynchronizeInvoke 2: { 3: // Methods 4: [HostProtection(SecurityAction.LinkDemand, Synchronization=true, ExternalThreading=true)] 5: IAsyncResult BeginInvoke(Delegate method, object[] args); 6: object EndInvoke(IAsyncResult result); 7: object Invoke(Delegate method, object[] args); 8:? 9: // Properties 10: bool InvokeRequired { get; } 11: } 12:?

如何我們放棄基于SynchonizationContext的解決方案，我們也可以通過基于ISynchronizeInvoke的方式來解決這個(gè)問題。為此我們這樣定義CalculateCallback：

1: namespace Artech.ThreadAffinity.Clients 2: { 3: [CallbackBehavior(UseSynchronizationContext = false)] 4: public class CalculateCallback : ICalculateCallback 5: { 6: public static TextBox ResultPanel; 7: public delegate void DisplayResultDelegate(TextBox resultPanel, double result); 8:? 9: #region ICalculateCallback Members 10:? 11: public void DisplayResult(double result) 12: { 13: DisplayResultDelegate displayResultDelegate = new DisplayResultDelegate(DisplayResult); 14: ResultPanel.BeginInvoke(displayResultDelegate, new object[] { ResultPanel, result }); 15: } 16:? 17: private void DisplayResult(TextBox resultPanel, double result) 18: { 19: resultPanel.Text = result.ToString(); 20: } 21:? 22: #endregion 23: } 24: } 25:?

由于BeginInvoke方式只能接受一個(gè)具體的delegate對(duì)象（不能使用匿名方法），所以需要定義一個(gè)具體的Delegate（DisplayResultDelegate）和對(duì)應(yīng)的方法（DisplayResult），參數(shù)通過一個(gè)object[]傳入。

從本質(zhì)上將，這兩種方式的實(shí)現(xiàn)完全是一樣的，如何你查看System.Windows.Forms.WindowsFormsSynchronizationContext的代碼，你會(huì)發(fā)現(xiàn)其Send和Post方方法就是通過調(diào)用Invoke和BeginInvoke方式實(shí)現(xiàn)的。

六、Service Hosting的線程關(guān)聯(lián)性

我們花了很多的精力介紹了WCF Duplex通信中Callback操作的線程關(guān)聯(lián)性問題，實(shí)際上我們使用到更多的還是service操作的線程關(guān)聯(lián)性問題。就以我們上面的程序?yàn)槔?#xff0c;我們通過一個(gè)Windows Form application來host我們的service，并且要求service的運(yùn)算結(jié)束后將結(jié)果輸出到server端的Window form的ListBox中，對(duì)ListBox的操作肯定需要的Host程序的UI線程中執(zhí)行。

按照我們一般的想法，我們的Service面向若干client，肯定是并發(fā)的接收client端的請(qǐng)求，以多線程的方式執(zhí)行service的操作，那么操作中UI 控件的操作肯定會(huì)出現(xiàn)錯(cuò)誤。

我們的程序依然可以正常運(yùn)行，其根本原因是WCF的service操作默認(rèn)實(shí)現(xiàn)了對(duì)Host service的當(dāng)前線程的SynchonizationContext實(shí)現(xiàn)了關(guān)聯(lián)。與Callback操作的線程關(guān)聯(lián)性通過CallbackBehavior的UseSynchronizationContext 進(jìn)行控制一樣，service的線程關(guān)聯(lián)性通過ServiceBehavir的UseSynchronizationContext 進(jìn)行設(shè)定。UseSynchronizationContext 的默認(rèn)值為true。

如何我們將CalculateService的UseSynchronizationContext 設(shè)為false：

1: namespace Artech.ThreadAffinity.Services 2: { 3: [ServiceBehavior(ConcurrencyMode = ConcurrencyMode.Reentrant,UseSynchronizationContext = false)] 4: public class CalculateService:ICalculate 5: { 6: public static ListBox DisplayPanel 7: { get; set; } 8:? 9: #region ICalculate Members 10:? 11: public void Add(double op1, double op2) 12: { 13: double result = op1 + op2; 14: ICalculateCallback callback = OperationContext.Current.GetCallbackChannel<ICalculateCallback>(); 15:? 16: DisplayPanel.Items.Add(string.Format("{0} + {1} = {2}", op1, op2, result)); 17:? 18: callback.DisplayResult(result); 19: } 20:? 21: #endregion 22: } 23: } 24:?

有control被不是創(chuàng)建它的線程操作，肯定會(huì)拋出一個(gè)InvalidOperationException，就像這樣：

我們一樣可以通過SynchonizationContext或者ISynchronizeInvoke的方式來解決這樣的問題，我們只討論前面一種，為此我們改變了CalculateService的定義：通過SynchonizationContext的Post方法實(shí)現(xiàn)對(duì)ListBox的訪問。

1: namespace Artech.ThreadAffinity.Services 2: { 3: [ServiceBehavior(ConcurrencyMode = ConcurrencyMode.Reentrant,UseSynchronizationContext = false)] 4: public class CalculateService:ICalculate 5: { 6: public static ListBox DisplayPanel 7: { get; set; } 8:? 9: public static SynchronizationContext SynchronizationContext 10: { get; set; } 11:? 12: #region ICalculate Members 13:? 14: public void Add(double op1, double op2) 15: { 16: double result = op1 + op2; 17: ICalculateCallback callback = OperationContext.Current.GetCallbackChannel<ICalculateCallback>(); 18: SynchronizationContext.Post(delegate 19: { 20: DisplayPanel.Items.Add(string.Format("{0} + {1} = {2}", op1, op2, result)); 21: }, null); 22:? 23: callback.DisplayResult(result); 24: } 25:? 26: #endregion 27: } 28: } 29:?

通過static屬性定義的SynchonizationContext在host的時(shí)候指定：

這樣我們的程序又可以正常運(yùn)行了。

WCF后續(xù)之旅：
WCF后續(xù)之旅(1): WCF是如何通過Binding進(jìn)行通信的
WCF后續(xù)之旅(2): 如何對(duì)Channel Layer進(jìn)行擴(kuò)展——?jiǎng)?chuàng)建自定義Channel
WCF后續(xù)之旅(3): WCF Service Mode Layer 的中樞—Dispatcher
WCF后續(xù)之旅(4)：WCF Extension Point 概覽
WCF后續(xù)之旅(5): 通過WCF Extension實(shí)現(xiàn)Localization
WCF后續(xù)之旅(6): 通過WCF Extension實(shí)現(xiàn)Context信息的傳遞
WCF后續(xù)之旅(7)：通過WCF Extension實(shí)現(xiàn)和Enterprise Library Unity Container的集成
WCF后續(xù)之旅(8)：通過WCF Extension 實(shí)現(xiàn)與MS Enterprise Library Policy Injection Application Block 的集成
WCF后續(xù)之旅(9)：通過WCF的雙向通信實(shí)現(xiàn)Session管理[Part I]
WCF后續(xù)之旅(9): 通過WCF雙向通信實(shí)現(xiàn)Session管理[Part II]
WCF后續(xù)之旅(10): 通過WCF Extension實(shí)現(xiàn)以對(duì)象池的方式創(chuàng)建Service Instance
WCF后續(xù)之旅(11): 關(guān)于并發(fā)、回調(diào)的線程關(guān)聯(lián)性（Thread Affinity）
WCF后續(xù)之旅(12): 線程關(guān)聯(lián)性（Thread Affinity）對(duì)WCF并發(fā)訪問的影響
WCF后續(xù)之旅(13): 創(chuàng)建一個(gè)簡單的WCF SOAP Message攔截、轉(zhuǎn)發(fā)工具[上篇]
WCF后續(xù)之旅(13)：創(chuàng)建一個(gè)簡單的SOAP Message攔截、轉(zhuǎn)發(fā)工具[下篇]
WCF后續(xù)之旅(14)：TCP端口共享
WCF后續(xù)之旅(15): 邏輯地址和物理地址
WCF后續(xù)之旅(16): 消息是如何分發(fā)到Endpoint的--消息篩選(Message Filter)
WCF后續(xù)之旅(17)：通過tcpTracer進(jìn)行消息的路由

轉(zhuǎn)載于:https://www.cnblogs.com/artech/archive/2008/08/21/1273021.html

總結(jié)

以上是生活随笔為你收集整理的WCF后续之旅(11): 关于并发、回调的线程关联性（Thread Affinity）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。