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

编程问答

我的WCF之旅(13)：创建基于MSMQ的Responsive Service

發(fā)布時間：2024/4/11 编程问答 30 豆豆

生活随笔收集整理的這篇文章主要介紹了我的WCF之旅(13)：创建基于MSMQ的Responsive Service 小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

一、One-way MEP V.S. Responsible Service

我們知道MSMQ天生就具有異步的特性，它只能以O(shè)ne-way的MEP（Message Exchange Pattern）進(jìn)行通信。Client和Service之間采用One-way MEP的話就意味著Client調(diào)用Service之后立即返回，它無法獲得Service的執(zhí)行結(jié)果，也無法捕捉Service運行的Exception。下圖簡單表述了基于MSMQ的WCF Service中Client和Service的交互。

但是在有些場景中，這是無法容忍的。再拿我在上一篇文章的Order Delivery的例子來說。Client向Service提交了Order，卻無法確認(rèn)該Order是否被Service正確處理，這顯然是不能接受的。我們今天就來討論一下，如何創(chuàng)建一個Responsive Service來解決這個問題：Client不再是對Service的執(zhí)行情況一無所知，它可以獲知Order是否被Service正確處理了。

二、?Solution

雖然我們的目的很簡單：當(dāng)Client向Service遞交了Order之后，能以某種方式獲知Order的執(zhí)行結(jié)果；對于Service端來說，在正確把Order從Message Queue中獲取出來、并正確處理之后，能夠向Order的遞交者發(fā)送一個Acknowledge Message。為了簡單起見，這個Acknowledge Message包含兩組信息：

Order No.: 被處理的Order的一個能夠為一標(biāo)志它的ID。
Exception: 如果處理失敗的Exception，如果成功處理為null。

要在WCF中實現(xiàn)這樣的目的，對于Request/Reply MEP來說是簡單而直接的：Client向Service遞交Order，并等待Service的Response，Service在處理接收到Order之后直接將處理結(jié)果返回給Client就可以了。但是我們說過MSMQ天生就是異步的，我們只有采取一種間接的方式實現(xiàn)“曲線救國”。

我們的解決方案是：在每個Client Domain也創(chuàng)建一個基于MSMQ的本地的WCF Service，用于接收來自O(shè)rder處理端發(fā)送的Acknowledge Message。對于處理Order 的Service來說，在正確處理Order之后，想對應(yīng)的Client發(fā)送Acknowledge Message。下圖簡單演示整個過程：

三、Implementation

了解了上面的Solution之后，我們來看看該Solution在真正實現(xiàn)過程中有什么樣的困難。對于處理Order的Service來說，在向Client端發(fā)送Acknowledge Message的時候，它必須要知道該Order對應(yīng)的Client的Response Service的MSMQ的Address以及其他和Operation相關(guān)的Context信息（在這里我們不需要，不過考慮到擴(kuò)展性，我們把包括了address的Context的信息封裝到一個了Class中，在這里叫做：OrderResponseContext）。而這些Context卻不能在Configuration中進(jìn)行配置，因為他可以同時面臨著很多個Client：比如每個Client用于接收Response 的Message Queue的address都不一樣。所以這個OrderResponseContext必須通過對應(yīng)的Client來提供。基于此，我們具有兩面兩種解決方式：

方式一、修改Service Contract，把OrderResponseContext當(dāng)成是Operation的一個參數(shù)

這是我們最容易想到的，比如我們原來的Operation這樣定義：

namespace?Artech.ResponsiveQueuedService.Contract
{
????[ServiceContract]
????[ServiceKnownType(typeof(Order))]
????public?interface?IOrderProcessor
????{
????????[OperationContract(IsOneWay?=?true)]
????????void?Submit(Order?order);
????}
}

現(xiàn)在變成：

雖然這種方式看起來不錯，但是卻不值得推薦。在一般情況下，我們的Contract需要是很穩(wěn)定的，一經(jīng)確定就不能輕易更改，因為Contract是被交互的多方共同支持的，牽一發(fā)動全身；此外，從Service Contract代表的是Service的一個Interface，他是對業(yè)務(wù)邏輯的抽象、和具體實現(xiàn)無關(guān)，而對于我們的例子來說，我們僅僅是定義一個遞交Order的Operation，從業(yè)務(wù)邏輯來看，OrderResponseContext和抽象的業(yè)務(wù)邏輯毫無關(guān)系。基于此，我們需要尋求一種和Service Contract無關(guān)的解決方式：

方式二、將OrderResponseContext放到Soap Message 的Header中

其實我們要解決的問題很簡單，就是要把OrderResponseContext的信息置于Soap Message中發(fā)送到Service。而我們知道，Soap的Header具有極強的可伸縮性，原則上，我們可以把任何控制信息置于Header中。基于WCF的編程模式很容易地幫助我們實現(xiàn)對Soap Header的插入和獲取：

我們可以通過下面的方式獲得當(dāng)前Operation Context的Incoming Message Headers和Outgoing Message Headers

OperationContext.Current.IncomingMessageHeaders
OperationContext.Current.OutgoingMessageHeaders

如果我們要把一個OrderResponseContext 對象插入到當(dāng)前Operation Context的Outgoing Message Headers中，我們可以通過下面的代碼來實現(xiàn)：

OrderResponseContext?context?=?new?OrderResponseContext();
MessageHeader<OrderResponseContext>?header?=?new?MessageHeader<OrderResponseContext>(?context);
OperationContext.Current.OutgoingMessageHeaders.Add(header.GetUntypedHeader("name",?"namespace"));

相應(yīng)的，我們可以通過下面的代碼從Outgoing Message Headers OrderResponseContext的數(shù)據(jù)獲取的內(nèi)容：

OrderResponseContext?context?=?OperationContext.Current.IncomingMessageHeaders.GetHeader<OrderResponseContext>("name",?"namespace"));

四、Sample

我們照例給出一個完整的Sample，下面是整個Solution的結(jié)構(gòu)：

除了一貫使用的4層結(jié)構(gòu)（Contract-Service-Hosting-Client），還為ResponseService增加了下面兩層：

Localservice: 作為Client Domain的ResponseService。
LocalHosting：Host Localservice。

1．Contract: ?Artech.ResponsiveQueuedService.Contract

Service Contract: Artech.ResponsiveQueuedService.Contract. IOrderProcessor

using?System;
using?System.Collections.Generic;
using?System.Text;
using?System.ServiceModel;

namespace?Artech.ResponsiveQueuedService.Contract
{
????[ServiceContract]
????[ServiceKnownType(typeof(Order))]
????public?interface?IOrderProcessor
????{
????????[OperationContract(IsOneWay?=?true)]
????????void?Submit(Order?order);
????}
}

Service Contract: Artech.ResponsiveQueuedService.Contract.IOrderRessponse

using?System;
using?System.Collections.Generic;
using?System.Text;
using?System.ServiceModel;

namespace?Artech.ResponsiveQueuedService.Contract
{
????[ServiceContract]
????public?interface??IOrderRessponse
????{
????????[OperationContract(IsOneWay?=true)]
????????void?SubmitOrderResponse(Guid?orderNo,FaultException?exception);
????}
}

接收來自O(shè)rder processing端的Response：Order No.和Exception。

Data Contract: Artech.ResponsiveQueuedService.Contract.Order

using?System;
using?System.Collections.Generic;
using?System.Text;
using?System.Runtime.Serialization;

namespace?Artech.ResponsiveQueuedService.Contract
{
????[DataContract]
????public?class?Order
????{
????????Private?Fields#region?Private?Fields
????????private?Guid?_orderNo;
????????private?DateTime?_orderDate;
????????private?Guid?_supplierID;
????????private?string?_supplierName;
????????#endregion

????????Constructors#region?Constructors
????????public?Order(Guid?orderNo,?DateTime?orderDate,?Guid?supplierID,?string?supplierName)
????????{
????????????this._orderNo?=?orderNo;
????????????this._orderDate?=?orderDate;
????????????this._supplierID?=?supplierID;
????????????this._supplierName?=?supplierName;
????????}

????????#endregion

????????Public?Properties#region?Public?Properties
????????[DataMember]
????????public?Guid?OrderNo
????????{
????????????get?{?return?_orderNo;?}
????????????set?{?_orderNo?=?value;?}
????????}

????????[DataMember]
????????public?DateTime?OrderDate
????????{
????????????get?{?return?_orderDate;?}
????????????set?{?_orderDate?=?value;?}
????????}

????????[DataMember]
????????public?Guid?SupplierID
????????{
????????????get?{?return?_supplierID;?}
????????????set?{?_supplierID?=?value;?}
????????}

????????[DataMember]
????????public?string?SupplierName
????????{
????????????get?{?return?_supplierName;?}
????????????set?{?_supplierName?=?value;?}
????????}
????????#endregion

????????Public?Methods#region?Public?Methods
????????public?override?string?ToString()
????????{
????????????string?description?=?string.Format("Order?No./t:?{0}/n/tOrder?Date/t:?{1}/n/tSupplier?No./t:?{2}/n/tSupplier?Name/t:?{3}",?
????????????????this._orderNo,?this._orderDate.ToString("yyyy/MM/dd"),?this._supplierID,?this._supplierName);
????????????return?description;
????????}
????????#endregion
????}
}

對Order的封裝。

Data Contract：Artech.ResponsiveQueuedService.Contract. OrderResponseContext

using?System;
using?System.Collections.Generic;
using?System.Text;
using?System.Runtime.Serialization;
using?System.ServiceModel;

namespace?Artech.ResponsiveQueuedService.Contract
{????
????[DataContract]
????public?class?OrderResponseContext
????{
????????private?Uri?_responseAddress;

????????[DataMember]
????????public?Uri?ResponseAddress
????????{
????????????get?{?return?_responseAddress;?}
????????????set?{?_responseAddress?=?value;?}
????????}

????????public?static?OrderResponseContext?Current
????????{
????????????get
????????????{
????????????????if?(OperationContext.Current?==?null)
????????????????{
????????????????????return?null;
????????????????}

????????????????return?OperationContext.Current.IncomingMessageHeaders.GetHeader<OrderResponseContext>("OrderResponseContext",?"Artech.ResponsiveQueuedService.Contract");
????????????}
????????????set
????????????{
????????????????MessageHeader<OrderResponseContext>?header?=?new?MessageHeader<OrderResponseContext>(value);
????????????????OperationContext.Current.OutgoingMessageHeaders.Add(header.GetUntypedHeader("OrderResponseContext",?"Artech.ResponsiveQueuedService.Contract"));
????????????}
????????}
????}
}

ResponseAddress代表Host在Client Domain的Response Service的Address。同過Current把OrderResponseContext插入到Outgoing Message Headers中、以及從Ingoing Message Headers取出OrderResponseContext對象。

2．Order Processing Service：Artech.ResponsiveQueuedService.Service

using?System;
using?System.Collections.Generic;
using?System.Text;
using?Artech.ResponsiveQueuedService.Contract;
using?System.ServiceModel;
using?System.Net.Security;

namespace?Artech.ResponsiveQueuedService.Service
{
????public?class?OrderProcessorService:IOrderProcessor
????{
????????private?void?ProcessOrder(Order?order)
????????{

????????????if?(order.OrderDate?<?DateTime.Today)
????????????{
????????????????throw?new?Exception();
????????????}
????????}

????????IOrderProcessor?Members#region?IOrderProcessor?Members

????????public?void?Submit(Order?order)
????????{
????????????Console.WriteLine("Begin?to?process?the?order?of?the?order?No.:?{0}",?order.OrderNo);
????????????FaultException?exception=?null;
????????????if?(order.OrderDate?<?DateTime.Today)
????????????{
????????????????exception?=?new?FaultException(new?FaultReason("The?order?has?expried"),?new?FaultCode("sender"));
????????????????Console.WriteLine("It's?fail?to?process?the?order./n/tOrder?No.:?{0}/n/tReason:{1}",?order.OrderNo,?"The?order?has?expried");
????????????}
????????????else
????????????{
????????????????Console.WriteLine("It's?successful?to?process?the?order./n/tOrder?No.:?{0}",?order.OrderNo);
????????????}

????????????NetMsmqBinding?binding?=?new?NetMsmqBinding();
????????????binding.ExactlyOnce?=?false;
????????????binding.Security.Transport.MsmqAuthenticationMode?=?MsmqAuthenticationMode.None;
????????????binding.Security.Transport.MsmqProtectionLevel?=?ProtectionLevel.None;
????????????ChannelFactory<IOrderRessponse>?channelFacotry?=?new?ChannelFactory<IOrderRessponse>(binding);
????????????OrderResponseContext?responseContext?=?OrderResponseContext.Current;
????????????IOrderRessponse?channel?=?channelFacotry.CreateChannel(new?EndpointAddress(responseContext.ResponseAddress));

????????????using?(OperationContextScope?contextScope?=?new?OperationContextScope(channel?as?IContextChannel))
????????????{
????????????????channel.SubmitOrderResponse(order.OrderNo,?exception);
????????????}
????????}
????????#endregion
????}
}

在這里我們模擬了這樣的場景：先通過Order Date判斷Order是否過期，如果過期創(chuàng)建一個FaultException，否則正確處理該Order，然后通過OrderResponseContext.Current從Incoming Message Header中獲取封裝在OrderResponseContext對象中的Response Address，創(chuàng)建Binding并調(diào)用Response Service.

3. Order Processing Service Hosting: Artech.ResponsiveQueuedService.Hosting

Configuration

<?xml?version="1.0"?encoding="utf-8"??>
<configuration>
??<appSettings>
????<add?key="msmqPath"?value="./private$/orderprocessor"/>
??</appSettings>
??<system.serviceModel>
????<bindings>
??????<netMsmqBinding>
????????<binding?name="MsmqBinding"?exactlyOnce="false"?useActiveDirectory="false">
??????????<security>
????????????<transport?msmqAuthenticationMode="None"?msmqProtectionLevel="None"?/>
??????????</security>
????????</binding>
??????</netMsmqBinding>
????</bindings>
????<services>
??????<service?name="Artech.ResponsiveQueuedService.Service.OrderProcessorService">
????????<endpoint?address="net.msmq://localhost/private/orderprocessor"?binding="netMsmqBinding"
????????????bindingConfiguration="MsmqBinding"?contract="Artech.ResponsiveQueuedService.Contract.IOrderProcessor"?/>
??????</service>
????</services>
??</system.serviceModel>
</configuration>

Program

using?System;
using?System.Collections.Generic;
using?System.Text;
using?Artech.ResponsiveQueuedService.Service;
using?System.ServiceModel;
using?System.Configuration;
using?System.Messaging;

namespace?Artech.ResponsiveQueuedService.Hosting
{
????class?Program
????{
????????static?void?Main(string[]?args)
????????{
????????????string?path?=?ConfigurationManager.AppSettings["msmqPath"];
????????????if?(!MessageQueue.Exists(path))
????????????{
????????????????MessageQueue.Create(path);
????????????}

????????????using?(ServiceHost?host?=?new?ServiceHost(typeof(OrderProcessorService)))
????????????{
????????????????host.Opened?+=?delegate
????????????????{
????????????????????Console.WriteLine("The?Order?Processor?service?has?begun?to?listen");
????????????????};

????????????????host.Open();

????????????????Console.Read();
????????????}
????????}
????}
}

4. Response Service: Artech.ResponsiveQueuedService.LocalService.OrderRessponseService

using?System;
using?System.Collections.Generic;
using?System.Text;
using?Artech.ResponsiveQueuedService.Contract;
using?System.ServiceModel;

namespace?Artech.ResponsiveQueuedService.LocalService
{
????public?class?OrderRessponseService?:?IOrderRessponse
????{
????????IOrderRessponse?Members#region?IOrderRessponse?Members

????????public?void?SubmitOrderResponse(Guid?orderNo,?FaultException?exception)
????????{
????????????if?(exception?==?null)
????????????{
????????????????Console.WriteLine("It's?successful?to?process?the?order!/n/tOrder?No.:?{0}",orderNo);
????????????}
????????????else
????????????{
????????????????Console.WriteLine("It's?fail?to?process?the?order!/n/tOrder?No.:?{0}/n/tReason:?{1}",?orderNo,?exception.Message);
????????????}
????????}

????????#endregion
????}
}

5. Response Service Hosting: Artech.ResponsiveQueuedService.LocalhHosting

Configuration

<?xml?version="1.0"?encoding="utf-8"??>
<configuration>
??<appSettings>
????<add?key="msmqPath"?value="./private$/orderresponse"/>
??</appSettings>
??<system.serviceModel>
????<bindings>
??????<netMsmqBinding>
????????<binding?name="msmqBinding"?exactlyOnce="false">
??????????<security>
????????????<transport?msmqAuthenticationMode="None"?msmqProtectionLevel="None"?/>
??????????</security>
????????</binding>
??????</netMsmqBinding>
????</bindings>
????<services>
??????<service?name="Artech.ResponsiveQueuedService.LocalService.OrderRessponseService">
????????<endpoint?address="net.msmq://localhost/private/orderresponse"?binding="netMsmqBinding"
????????????bindingConfiguration="msmqBinding"?contract="Artech.ResponsiveQueuedService.Contract.IOrderRessponse"?/>
??????</service>
????</services>
??</system.serviceModel>
</configuration>

Program

using?System;
using?System.Collections.Generic;
using?System.Text;
using?Artech.ResponsiveQueuedService.LocalService;
using?System.Configuration;
using?System.ServiceModel;
using?System.Messaging;

namespace?Artech.ResponsiveQueuedService.LocalhHosting
{
????class?Program
????{
????????static?void?Main(string[]?args)
????????{
????????????string?path?=?ConfigurationManager.AppSettings["msmqPath"];
????????????if?(!MessageQueue.Exists(path))
????????????{
????????????????MessageQueue.Create(path);
????????????}

????????????using?(ServiceHost?host?=?new?ServiceHost(typeof(OrderRessponseService)))
????????????{
????????????????host.Opened?+=?delegate
????????????????{
????????????????????Console.WriteLine("The?Order?Response?service?has?begun?to?listen");
????????????????};

????????????????host.Open();

????????????????Console.Read();
????????????}
????????}
????}
}

6. Client: Artech.ResponsiveQueuedService.Client

Configuration:

<?xml?version="1.0"?encoding="utf-8"??>
<configuration>
??<appSettings>
????<add?key="msmqPath"?value="net.msmq://localhost/private/orderresponse"/>
??</appSettings>
??<system.serviceModel>
????<bindings>
??????<netMsmqBinding>
????????<binding?name="MsmqBinding"?exactlyOnce="false"?useActiveDirectory="false">
??????????<security>
????????????<transport?msmqAuthenticationMode="None"?msmqProtectionLevel="None"?/>
??????????</security>
????????</binding>
??????</netMsmqBinding>
????</bindings>
????<client>
??????<endpoint?address="net.msmq://localhost/private/orderprocessor"?binding="netMsmqBinding"
????????????bindingConfiguration="MsmqBinding"?contract="Artech.ResponsiveQueuedService.Contract.IOrderProcessor"?name="defaultEndpoint"?/>
????</client>
??</system.serviceModel>
</configuration>

Program:

using?System;
using?System.Collections.Generic;
using?System.Text;
using?System.Configuration;
using?System.ServiceModel;
using?Artech.ResponsiveQueuedService.Contract;
using?System.Messaging;

namespace?Artech.ResponsiveQueuedService.Clinet
{
????class?Program
????{
????????static?void?Main(string[]?args)
????????{
????????????Order?order1?=?new?Order(Guid.NewGuid(),?DateTime.Today.AddDays(5),?Guid.NewGuid(),?"Supplier?A");
????????????Order?order2?=?new?Order(Guid.NewGuid(),?DateTime.Today.AddDays(-5),?Guid.NewGuid(),?"Supplier?A");

????????????string?path?=?ConfigurationManager.AppSettings["msmqPath"];
????????????Uri?address?=?new?Uri(path);
????????????OrderResponseContext?context?=?new?OrderResponseContext();
????????????context.ResponseAddress?=?address;

????????????ChannelFactory<IOrderProcessor>?channelFactory?=?new?ChannelFactory<IOrderProcessor>("defaultEndpoint");
????????????IOrderProcessor?orderProcessor?=?channelFactory.CreateChannel();

????????????using?(OperationContextScope?contextScope?=?new?OperationContextScope(orderProcessor?as?IContextChannel))
????????????{
????????????????Console.WriteLine("Submit?the?order?of?order?No.:?{0}",?order1.OrderNo);
????????????????OrderResponseContext.Current?=?context;
????????????????orderProcessor.Submit(order1);
????????????}

????????????using?(OperationContextScope?contextScope?=?new?OperationContextScope(orderProcessor?as?IContextChannel))
????????????{
????????????????Console.WriteLine("Submit?the?order?of?order?No.:?{0}",?order2.OrderNo);
????????????????OrderResponseContext.Current?=?context;
????????????????orderProcessor.Submit(order2);
????????????}

????????????Console.Read();
????????}
????}
}

我創(chuàng)建了兩個Order對象, 其中一個已經(jīng)過期。從Configuration中取出Response Address并購建一個OrderResponseContext，然后分兩次將這兩個Order向Order Processing Service遞交。在調(diào)用Order Processing Order的Operation Context Scope中，通過OrderResponseContext.Current將OrderResponseContext對象插入Outcoming Message Header中。

我們現(xiàn)在運行一下整個程序，看看最終的輸出結(jié)果：

Client：

Order Processing：

Order Response：

Reference：
Build a Queued WCF Response Service

作者：Artech
出處：http://artech.cnblogs.com
本文版權(quán)歸作者和博客園共有，歡迎轉(zhuǎn)載，但未經(jīng)作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接，否則保留追究法律責(zé)任的權(quán)利。

總結(jié)

以上是生活随笔為你收集整理的我的WCF之旅(13)：创建基于MSMQ的Responsive Service的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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

上一篇： Windows Phone 7项目实战之
下一篇：一项调查结果：你的分析/挖掘计算机的配置