WCF分布式開發(fā)步步為贏(11):WCF流處理（Streaming）機(jī)制
WSE3.0框架提供了數(shù)據(jù)優(yōu)化傳輸機(jī)制，WSE3.0構(gòu)建Web服務(wù)安全(4):MTOM消息傳輸優(yōu)化和文件上傳、下載 疑問里進(jìn)行了介紹。WCF同樣也提供了流操作來支持大數(shù)據(jù)對象的傳輸和處理優(yōu)化機(jī)制，今天我們WCF分布式開發(fā)步步為贏系列的(4):使用流操作（Streaming Operations）優(yōu)化傳輸。本節(jié)會詳細(xì)介紹流操作的相關(guān)概念、編程實(shí)現(xiàn)過程，以及實(shí)際開發(fā)過程中需要主要的一些問題。本節(jié)結(jié)構(gòu)：【1】流處理的概念【2】流處理的特點(diǎn)【3】示例代碼分析【4】總結(jié)。最后上傳本文的示例代碼。
Streaming，本文翻譯為流處理（張逸兄翻譯的Programming WCF Services一書里把這個機(jī)制翻譯為“流操作”，不存在爭議。我選擇“流處理”一詞的意思，只是想把這個Streaming詞形象準(zhǔn)確化，動名詞，流化處理、流處理，翻譯為流操作，初學(xué)者會誤會認(rèn)為這個是一個服務(wù)操作。因?yàn)镾treaming只是WCF內(nèi)建的一個機(jī)制。而不是操作。）
我們首先來理解什么是Streaming流處理。
【1】Streaming流處理的概念：
通常情況，客戶端和服務(wù)端進(jìn)行交互，傳遞消息，都是放到接收端的緩存里，待接收完畢后再進(jìn)行處理。無論接收端是客戶端還是服務(wù)端都是如此。
【1.1】要解決的問題：
當(dāng)客戶端調(diào)用服務(wù)時，要阻塞客戶單進(jìn)程，直到消息發(fā)送完畢，服務(wù)端才開始處理數(shù)據(jù)，然后是返回處理完畢的結(jié)果給客戶端，客戶端接收完畢，才能解除阻塞。這樣帶來的問題是當(dāng)消息傳遞的時間很短，相對處理時間可以忽略不計，不會影響系統(tǒng)服務(wù)的效率。但是要是消息數(shù)據(jù)很大，比如是圖片或者多媒體對象。每次傳輸時間相對較大，這樣接收端的等待時間過久，勢必每次阻塞都會很長，進(jìn)程無法繼續(xù)執(zhí)行。因而導(dǎo)致效率低下。
【1.2】Streaming流處理：
Streaming流處理就是WCF提供的主要針對大量消息數(shù)據(jù)處理的一種優(yōu)化機(jī)制。WCF允許接收端通過通道接受消息的同時，啟動對消息數(shù)據(jù)的處理，這樣的過程稱為流傳輸模型。
【2】Streaming流處理的特點(diǎn)：
顯然對于處理大量的消息數(shù)據(jù)而言，流處理機(jī)制改善了系統(tǒng)的吞吐量和響應(yīng)效率。
【2.1】流處理操作定義：
WCF的流處理機(jī)制需要使用.NET FrameWork定義的Stream類（它是FileStream, NetworkStream, MemoryStream 的父類）。流處理適用一下場景：
interface IMyContract
{
[OperationContract]
Stream StreamReply1( );
[OperationContract]
void StreamReply2(out Stream stream);
[OperationContract]
void StreamRequest(Stream stream);
[OperationContract(IsOneWay = true)]
void OneWayStream(Stream stream);
}
它可以做為返回數(shù)據(jù)、參數(shù)、輸出參數(shù)的類型。當(dāng)然也可以作為單調(diào)服務(wù)的操作參數(shù)。這里使用的參數(shù)必須是可序列化的，例如MemoryStream。而FileStream不支持序列化因而不能作為參數(shù)或者返回數(shù)據(jù)的類型。
【2.2】流處理與綁定協(xié)議：
流處理機(jī)制在特定的綁定協(xié)議中才能使用，目前是BasicHttpBinding, NetTcpBinding, 和NetNamedPipeBinding 支持流處理模型。但是在默認(rèn)情況下，WCF禁止流處理模式。
流傳輸模式使用使用TransferMode進(jìn)行配置，TransferMode為枚舉類型，其定義如下：
public enum TransferMode
{
// Summary:
// The request and response messages are both buffered.
Buffered = 0,
//
// Summary:
// The request and response messages are both streamed.
Streamed = 1,
//
// Summary:
// The request message is streamed and the response message is buffered.
StreamedRequest = 2,
//
// Summary:
// The request message is buffered and the response message is streamed.
StreamedResponse = 3,
}
只有Streamed模式支持2.1中列舉的流處理模式場景。除了直接在服務(wù)上配置屬性以外，我們還可以再服務(wù)的配置文件里定義流傳輸模式。代碼如下：
<binding name="basicHttpBinding" receiveTimeout="10:10:10" transferMode="Streamed" maxReceivedMessageSize="200000">
</binding>
</basicHttpBinding>
<netTcpBinding>
<binding name="netTcpBinding" receiveTimeout="10:10:10" transferMode="Streamed" maxReceivedMessageSize="200000">
</binding>
</netTcpBinding>
此為托管宿主的配置文件，特定的綁定協(xié)議，可以配置其傳輸模式。
【2.3】注意：
流處理在使用http協(xié)議時，其默認(rèn)消息長度是64K,如果希望增加數(shù)據(jù)長度，需要在配置文件里重新設(shè)置。如： maxReceivedMessageSize="200000"，具體代碼如下：
<binding name="basicHttpBinding" receiveTimeout="10:10:10" transferMode="Streamed" maxReceivedMessageSize="200000">
</binding>
</basicHttpBinding>
【3】示例代碼分析：
這里測試的流處理機(jī)制，使用的是處理圖片的上傳于下載，分別使用Stream和其子類MemoryStream作為參數(shù)或者返回消息數(shù)據(jù)的類型。基本代碼演示的是流處理三種模式場景：
【3.1】服務(wù)端：
服務(wù)契約分別定義了下載數(shù)據(jù)和上傳數(shù)據(jù)，下載數(shù)據(jù)使用的類型MemoryStream，上傳數(shù)據(jù)參數(shù)類型是是Stream。具體代碼如下：
[ServiceContract( Namespace = "http://www.rzrgm.cn/frank_xl/")]
public interface IWCFService
{
//操作契約,獲取數(shù)據(jù)流
[OperationContract]
MemoryStream DownLoadStreamData(string fileName);
//操作契約，輸出數(shù)據(jù)流
[OperationContract]
void DownLoadStreamDataOut(out MemoryStream stream, string fileName);
//操作契約，上載數(shù)據(jù)流,單向操作的消息轉(zhuǎn)換為數(shù)據(jù)流
[OperationContract(IsOneWay=true)]
void UpLoadStreamData(Stream stream);
}
//2.服務(wù)類，繼承接口。實(shí)現(xiàn)服務(wù)契約定義的操作
public class WCFService : IWCFService
{
//1實(shí)現(xiàn)接口定義的方法，下載文件數(shù)據(jù)流
public MemoryStream DownLoadStreamData(string fileName)
{
// Stream stream =
byte[] file = new byte[200000];
String filePath = AppDomain.CurrentDomain.BaseDirectory + @"/" + fileName;
file = File.ReadAllBytes(filePath);
MemoryStream memoryStream = new MemoryStream(file);
return memoryStream;
}
//2實(shí)現(xiàn)接口定義的方法，下載文件數(shù)據(jù)流
public void DownLoadStreamDataOut(out MemoryStream stream, string fileName)
{
// Stream stream =
byte[] file = new byte[200000];
String filePath = AppDomain.CurrentDomain.BaseDirectory + @"/" + fileName;
file = File.ReadAllBytes(filePath);
MemoryStream memoryStream = new MemoryStream(file);
stream = memoryStream;
}
//3實(shí)現(xiàn)接口定義的方法，上傳文件數(shù)據(jù)流
public void UpLoadStreamData(Stream stream)
{
// codes here to deal with the stream Stream stream =
Console.WriteLine("The Stream length is {0}",stream.Length);
}
}
【3.2】托管宿主：
我們分別使用basicHttpBinding和netTcpBinding定義了兩個服務(wù)終結(jié)點(diǎn)，這里不要忘記設(shè)置最大接受消息數(shù)據(jù)大小maxReceivedMessageSize="200000"，如果設(shè)置較小會導(dǎo)致接受數(shù)據(jù)超過設(shè)定的錯誤。具體代碼如下：
<services>
<service behaviorConfiguration="WCFService.WCFServiceBehavior"
name="WCFService.WCFService">
<endpoint
address="http://localhost:8002/WCFService"
binding="basicHttpBinding"
contract="WCFService.IWCFService">
</endpoint>
<endpoint
address="net.tcp://localhost:8004/WCFService"
binding="netTcpBinding"
contract="WCFService.IWCFService">
</endpoint>
<endpoint address="mex" binding="mexHttpBinding" contract="IMetadataExchange" />
<endpoint address="mex" binding="mexTcpBinding" contract="IMetadataExchange" />
<host>
<baseAddresses>
<add baseAddress="http://localhost:8001/"/>
<add baseAddress="net.tcp://localhost:8003/"/>
</baseAddresses>
</host>
</service>
</services>
<behaviors>
<serviceBehaviors>
<behavior name="WCFService.WCFServiceBehavior">
<serviceMetadata httpGetEnabled="true" />
<serviceDebug includeExceptionDetailInFaults="false" />
</behavior>
</serviceBehaviors>
</behaviors>
<bindings>
<basicHttpBinding>
<binding name="basicHttpBinding" receiveTimeout="10:10:10" transferMode="Streamed" maxReceivedMessageSize="200000">
</binding>
</basicHttpBinding>
<netTcpBinding>
<binding name="netTcpBinding" receiveTimeout="10:10:10" transferMode="Streamed" maxReceivedMessageSize="200000">
</binding>
</netTcpBinding>
</bindings>
</system.serviceModel>
【3.3】客戶端：
客戶端分別測試數(shù)據(jù)的上傳和下載，使用不同的方法。這里的測試目前
【4】總結(jié)：
本文介紹了WCF流處理模型，但是實(shí)現(xiàn)代碼出現(xiàn)問題。
（1）作為WCF的流處理機(jī)制，確實(shí)為我們的大規(guī)模消息數(shù)據(jù)傳輸提供了理想的機(jī)制，提高了系統(tǒng)的效率和響應(yīng)速度。
（2）Stream作為.net類庫的內(nèi)部類型，在.net平臺上使用來說較為方便，但是與異構(gòu)平臺的交互勢必受到諸多限制，也違背了WCF跨平臺的初衷。
（3） 我在調(diào)試這個示例代碼的過程中遇到了幾個錯誤，基本都整理出來放到WCF分布式開發(fā)常見錯誤里了。使用netTcpBinding綁定進(jìn)行數(shù)據(jù)傳輸?shù)臅r候，一個很有價值的錯誤就是：the socket connection was aborted. this could be caused by an error processing your message or a receive timeout being exceeded by the remote host ...這個錯誤我google國內(nèi)和國外的一些資料，但是解決的辦法都是更換協(xié)議。其實(shí)是一種很無奈的措施，目前我還沒有找出更好的解決辦法就是基于netTcpBinding。表面的原因是服務(wù)處理超時。但是具體的錯誤信息沒有這樣簡單。我更換協(xié)議以后其他的流服務(wù)操作調(diào)用出了問題。所以這個只能針對特定的操作有幫助。我把這個錯誤收集起來供大家參考。也希望發(fā)揮大家的作用把這個問題解決。
也許這個錯誤應(yīng)該反映給WCF的開發(fā)小組，無論國內(nèi)還是國外的技術(shù)論壇，包括MSDN都有人遇到這樣的問題，而沒有一個最佳的解決方案。這里我對流處理示例代碼分別打包，目前都有問題。
<1>流處理機(jī)制示例1里代碼測試上傳文件，成功，下載文件錯誤。
/Files/frank_xl/WCFServiceStreamingFrankXuLei.rar
<2>這里使用的是字節(jié)數(shù)組，測試下載文件,下載文件是成功的，上傳文件失敗。/Files/frank_xl/WCFServiceStreamingByteArrayFrankXuLei.rar
兩個失敗的原因都是一樣，套接字中斷，連接超時。WCF分布式開發(fā)常見錯誤解決(10)：套接字連接中斷，The socket connection was aborted ，我進(jìn)行了整理，也查找了國外的論壇，沒有找到理想的解決辦法。我已經(jīng)嘗試了修改接受時間的限制，但是不起作用。我會繼續(xù)關(guān)注這個問題，也希望有興趣的朋友補(bǔ)充。MSDN論壇上有人提供了解決的方法，但是不理想，更換協(xié)議。希望微軟WCF的開發(fā)、測試小組早日注意這個問題。
<3>Streaming代碼修改完畢，測試成功,請大家下載新的文件參考：/Files/frank_xl/WCFServiceStreamingSuccefullyFrankXuLei.rar
以前代碼因?yàn)榕渲缅e誤，綁定的配置沒應(yīng)用對。我今天修改完畢。
參考文章：
1.《Programming WCF Services》；
浙公網(wǎng)安備 33010602011771號