分布式處理在大型企業(yè)應(yīng)用系統(tǒng)中，最大的優(yōu)勢是將負(fù)載分布。通過多臺服務(wù)器處理多個任務(wù)，以優(yōu)化整個系統(tǒng)的處理能力和運行效率。分布式處理的技術(shù)核心是完 成服務(wù)與服務(wù)之間、服務(wù)端與客戶端之間的通信。在.Net 1.1中，可以利用Web Service或者.Net Remoting來實現(xiàn)服務(wù)進(jìn)程之間的通信。本文將介紹一種基于消息的分布式處理架構(gòu)，利用了.Net Remoting技術(shù)，并參考了CORBA Naming Service的處理方式，且定義了一套消息體制，來實現(xiàn)分布式處理。?

?一、消息的定義

?????? 要實現(xiàn)進(jìn)程間的通信，則通信內(nèi)容的載體——消息，就必須在服務(wù)兩端具有統(tǒng)一的消息標(biāo)準(zhǔn)定義。從通信的角度來看，消息可以分為兩類：Request Messge和Reply Message。為簡便起見，這兩類消息可以采用同樣的結(jié)構(gòu)。

?????? 消息的主體包括ID，Name和Body，我們可以定義如下的接口方法，來獲得消息主體的相關(guān)屬性：

C#語言:? public interface IMessage:ICloneable
{
???? IMessageItemSequence GetMessageBody();
???? string GetMessageID();
???? string GetMessageName();

???? void SetMessageBody(IMessageItemSequence aMessageBody);
???? void SetMessageID(string aID);
???? void SetMessageName(string aName);
}

??? 消息主體類Message實現(xiàn)了IMessage接口。在該類中，消息體Body為IMessageItemSequence類型。這個類型用于Get和Set消息的內(nèi)容：Value和Item：

C#語言:? public interface IMessageItemSequence:ICloneable
{??????

???? IMessageItem GetItem(string aName);
???? void SetItem(string aName,IMessageItem aMessageItem);???????

???? string GetValue(string aName);??
???? void SetValue(string aName,string aValue);
}

?????? Value為string類型，并利用HashTable來存儲Key和Value的鍵值對。而Item則為IMessageItem類型，同樣的在 IMessageItemSequence的實現(xiàn)類中，利用HashTable存儲了Key和Item的鍵值對。

?????? IMessageItem支持了消息體的嵌套。它包含了兩部分：SubValue和SubItem。實現(xiàn)的方式和IMessageItemSequence相似。定義這樣的嵌套結(jié)構(gòu)，使得消息的擴(kuò)展成為可能。一般的結(jié)構(gòu)如下：

?????? IMessage——Name
???????????????????? ——ID
???????????????????? ——Body（IMessageItemSequence）
??????????????????????????? ——Value
??????????????????????????? ——Item（IMessageItem）
?????????????????????????????????? ——SubValue
?????????????????????????????????? ——SubItem（IMessageItem）
????????????????????????????????????????? ——……

?????? 各個消息對象之間的關(guān)系如下：

?

?????? 在實現(xiàn)服務(wù)進(jìn)程通信之前，我們必須定義好各個服務(wù)或各個業(yè)務(wù)的消息格式。通過消息體的方法在服務(wù)的一端設(shè)置消息的值，然后發(fā)送，并在服務(wù)的另一端獲得這些值。例如發(fā)送消息端定義如下的消息體：

C#語言: IMessageFactory factory = new MessageFactory();
IMessageItemSequence body = factory.CreateMessageItemSequence();
body.SetValue("name1","value1");
body.SetValue("name2","value2");

IMessageItem item = factory.CreateMessageItem();
item.SetSubValue("subname1","subvalue1");
item.SetSubValue("subname2","subvalue2");

IMessageItem subItem1 = factory.CreateMessageItem();
subItem1.SetSubValue("subsubname11","subsubvalue11");
subItem1.SetSubValue("subsubname12","subsubvalue12");
IMessageItem subItem2 = factory.CreateMessageItem();
subItem1.SetSubValue("subsubname21","subsubvalue21");
subItem1.SetSubValue("subsubname22","subsubvalue22");

item.SetSubItem("subitem1",subItem1);
item.SetSubItem("subitem2",subItem2);

body.SetItem("item",item);

//Send Request Message
MyServiceClient service = new MyServiceClient("Client");
IMessageItemSequence reply = service.SendRequest("TestService","Test1",body);

?????? 在接收消息端就可以通過獲得body的消息體內(nèi)容，進(jìn)行相關(guān)業(yè)務(wù)的處理。?

?二、.Net Remoting服務(wù)

?????? 在.Net中要實現(xiàn)進(jìn)程間的通信，主要是應(yīng)用Remoting技術(shù)。根據(jù)前面對消息的定義可知，實際上服務(wù)的實現(xiàn)，可以認(rèn)為是對消息的處理。因此，我們可以對服務(wù)進(jìn)行抽象，定義接口IService:

C#語言:? public interface IService
{
???? IMessage Execute(IMessage aMessage);
}

??????? Execute()方法接受一條Request Message，對其進(jìn)行處理后，返回一條Reply Message。在整個分布式處理架構(gòu)中，可以認(rèn)為所有的服務(wù)均實現(xiàn)該接口。但受到Remoting技術(shù)的限制，如果要實現(xiàn)服務(wù)，則該服務(wù)類必須繼承自 MarshalByRefObject，同時必須在服務(wù)端被Marshal。隨著服務(wù)類的增多，必然要在服務(wù)兩端都要對這些服務(wù)的信息進(jìn)行管理，這加大了 系統(tǒng)實現(xiàn)的難度與管理的開銷。如果我們從另外一個角度來分析服務(wù)的性質(zhì)，基于消息處理而言，所有服務(wù)均是對Request Message的處理。我們完全可以定義一個Request服務(wù)負(fù)責(zé)此消息的處理。

?????? 然而，Request服務(wù)處理消息的方式雖然一致，但畢竟服務(wù)實現(xiàn)的業(yè)務(wù)，即對消息處理的具體實現(xiàn)，卻是不相同的。對我們要實現(xiàn)的服務(wù)，可以分為兩大類： 業(yè)務(wù)服務(wù)與Request服務(wù)。實現(xiàn)的過程為：首先，具體的業(yè)務(wù)服務(wù)向Request服務(wù)發(fā)出Request請求，Request服務(wù)偵聽到該請求，然后 交由其偵聽的服務(wù)來具體處理。

?????? 業(yè)務(wù)服務(wù)均具有發(fā)出Request請求的能力，且這些服務(wù)均被Request服務(wù)所偵聽，因此我們可以為業(yè)務(wù)服務(wù)抽象出接口IListenService：

C#語言: public interface IListenService
{
???? IMessage OnRequest(IMessage aMessage);?
}

??????? Request服務(wù)實現(xiàn)了IService接口，并包含IListenService類型對象的委派，以執(zhí)行OnRequest()方法：

C#語言:? public class RequestListener:MarshalByRefObject,IService
{
???? public RequestListener(IListenService listenService)
???? {
???????? m_ListenService = listenService;
???? }

???? private IListenService m_ListenService;

???? #region IService Members

???? public IMessage Execute(IMessage aMessage)
???? {
???????? return this.m_ListenService.OnRequest(aMessage);
???? }??????

???? #endregion
???? public override object InitializeLifetimeService()
???? {
???????? return null;
???? }
}

?????? 在RequestListener服務(wù)中，繼承了MarshalByRefObject類，同時實現(xiàn)了IService接口。通過該類的構(gòu)造函數(shù)，接收IListService對象。

?????? 由于Request消息均由Request服務(wù)即RequestListener處理，因此，業(yè)務(wù)服務(wù)的類均應(yīng)包含一個RequestListener的 委派，唯一的區(qū)別是其服務(wù)名不相同。業(yè)務(wù)服務(wù)類實現(xiàn)IListenService接口，但不需要繼承MarshalByRefObject，因為被 Marshal的是該業(yè)務(wù)服務(wù)內(nèi)部的RequestListener對象，而非業(yè)務(wù)服務(wù)本身：

C#語言:? public abstract class Service:IListenService
{
???? public Service(string serviceName)
???? {
???????? m_ServiceName = serviceName;?
???????? m_RequestListener = new RequestListener(this);
???? }??????

???? #region IListenService Members
???? public IMessage OnRequest(IMessage aMessage)
???? {
???????? //……
???? }?

???? #endregion

???? private string m_ServiceName;
???? private RequestListener m_RequestListener;????
}

?????? Service類是一個抽象類，所有的業(yè)務(wù)服務(wù)均繼承自該類。最后的服務(wù)架構(gòu)如下：

?

?????? 我們還需要在Service類中定義發(fā)送Request消息的行為，通過它，才能使業(yè)務(wù)服務(wù)被RequestListener所偵聽。?

C#語言:? public IMessageItemSequence SendRequest(string aServiceName,string??????????????????????????????????????? aMessageName,IMessageItemSequence aMessageBody)
{

???? IMessage message = m_Factory.CreateMessage();
???? message.SetMessageName(aMessageName);
???? message.SetMessageID("");
???? message.SetMessageBody(aMessageBody);

???? IService service = FindService(aServiceName);
???? IMessageItemSequence replyBody = m_Factory.CreateMessageItemSequence();
???? if (service != null)
???? {
????????? IMessage replyMessage = service.Execute(message);
????????? replyBody = replyMessage.GetMessageBody();?????????
???? }
???? else
???? {?????????
????????? replyBody.SetValue("result","Failure");?????????
???? }
???? return replyBody;
}

?????? 注意SendRequest()方法的定義，其參數(shù)包括服務(wù)名，消息名和被發(fā)送的消息主體。而在實現(xiàn)中最關(guān)鍵的一點是FindService()方法。我 們要查找的服務(wù)正是與之對應(yīng)的RequestListener服務(wù)。不過，在此之前，我們還需要先將服務(wù)Marshal：

C#語言:? public void Initialize()
{?????????????????????????????????????????
??? RemotingServices.Marshal(this.m_RequestListener,this.m_ServiceName +? ".RequestListener");
}

?????? 我們Marshal的對象，是業(yè)務(wù)服務(wù)中的Request服務(wù)對象m_RequestListener，這個對象在Service的構(gòu)造函數(shù)中被實例化：

C#語言: m_RequestListener = new RequestListener(this);

?????? 注意，在實例化的時候是將this作為IListenService對象傳遞給RequestListener。因此，此時被Marshal的服務(wù)對象， 保留了業(yè)務(wù)服務(wù)本身即Service的指引。可以看出，在Service和RequestListener之間，采用了“雙重委派”的機(jī)制。

?????? 通過調(diào)用Initialize()方法，初始化了一個服務(wù)對象，其類型為RequestListener（或IService），其服務(wù)名 為：Service的服務(wù)名 + ".RequestListener"。而該服務(wù)正是我們在SendRequest()方法中要查找的Service：

C#語言: IService service = FindService(aServiceName);

?????? 下面我們來看看FindService()方法的實現(xiàn)：

C#語言:? protected IService FindService(string aServiceName)
{
??? lock (this.m_Services)
??? {
???????? IService service = (IService)m_Services[aServiceName];
???????? if (service != null)
???????? {
???????????? return service;
???????? }
???????? else
???????? {
???????????? IService tmpService = GetService(aServiceName);
???????????? AddService(aServiceName,tmpService);
???????????? return tmpService;
???????? }
??? }
}

??????? 可以看到，這個服務(wù)是被添加到m_Service對象中，該對象為SortedList類型，服務(wù)名為Key，IService對象為Value。如果沒有找到，則通過私有方法GetService()來獲得：

C#語言: private IService GetService(string aServiceName)
{
??? IService service = (IService)Activator.GetObject(typeof(RequestListener),
???????? "tcp://localhost:9090/" + aServiceName + ".RequestListener");
??? return service;
}

??????? 在這里，Channel、IP、Port應(yīng)該從配置文件中獲取，為簡便起見，這里直接賦為常量。

?????? 再分析SendRequest方法，在找到對應(yīng)的服務(wù)后，執(zhí)行了IService的Execute()方法。此時的IService為 RequestListener，而從前面對RequestListener的定義可知，Execute()方法執(zhí)行的其實是其偵聽的業(yè)務(wù)服務(wù)的 OnRequest()方法。

?????? 我們可以定義一個具體的業(yè)務(wù)服務(wù)類，來分析整個消息傳遞的過程。該類繼承于Service抽象類：

C#語言:? public class MyService:Service
{
???? public MyService(string aServiceName):base(aServiceName)
???? {}?????????
}

??????? 假設(shè)把進(jìn)程分為服務(wù)端和客戶端，那么對消息處理的步驟如下：
?1、 在客戶端調(diào)用MyService的SendRequest()方法發(fā)送Request消息；
?2、 查找被Marshal的服務(wù)，即RequestListener對象，此時該對象應(yīng)包含對應(yīng)的業(yè)務(wù)服務(wù)對象MyService；
?3、 在服務(wù)端調(diào)用RequestListener的Execute()方法。該方法則調(diào)用業(yè)務(wù)服務(wù)MyService的OnRequest()方法。

在這些步驟中，除了第一步在客戶端執(zhí)行外，其他的步驟均是在服務(wù)端進(jìn)行。

?三、業(yè)務(wù)服務(wù)對于消息的處理

?????? 前面實現(xiàn)的服務(wù)架構(gòu)，已經(jīng)較為完整地實現(xiàn)了分布式的服務(wù)處理。但目前的實現(xiàn)，并未體現(xiàn)對消息的處理。我認(rèn)為，對消息的處理，等價與具體的業(yè)務(wù)處理。這些業(yè) 務(wù)邏輯必然是在服務(wù)端完成。每個服務(wù)可能會處理單個業(yè)務(wù)，也可能會處理多個業(yè)務(wù)。并且，服務(wù)與服務(wù)之間仍然存在通信，某個服務(wù)在處理業(yè)務(wù)時，可能需要另一 個服務(wù)的業(yè)務(wù)行為。也就是說，每一種類的消息，處理的方式均有所不同，而這些消息的唯一標(biāo)識，則是在SendRequest()方法已經(jīng)有所體現(xiàn)的 aMessageName。

?????? 雖然，處理的消息不同，所需要的服務(wù)不同，但是根據(jù)我們對消息的定義，我們?nèi)匀豢梢詫⑦@些消息處理機(jī)制抽象為一個統(tǒng)一的格式；在.Net中，體現(xiàn)這種機(jī)制的莫過于委托delegate。我們可以定義這樣的一個委托：

C#語言:? public delegate void RequestHandler(string aMessageName,IMessageItemSequence aMessageBody,ref IMessageItemSequence aReplyMessageBody);

?????? 在RequestHandler委托中，它代表了這樣一族方法：接收三個入 參，aMessageName，aMessageBody，aReplyMessageBody，返回值為void。其中，aMessageName代表 了消息名，它是消息的唯一標(biāo)識；aMessageBody是待處理消息的主體，業(yè)務(wù)所需要的所有數(shù)據(jù)都存儲在aMessageBody對象中。 aReplyMessageBody是一個引用對象，它存儲了消息處理后的返回結(jié)果，通常情況下，我們可以 用<"result","Success">或<"result", "Failure">來代表處理的結(jié)果是成功還是失敗。

?????? 這些委托均在服務(wù)初始化時被添加到服務(wù)類的SortedList對象中，鍵值為aMessageName。所以我們可以在抽象類中定義如下方法：?????

C#語言:? protected abstract void AddRequestHandlers();
protected void AddRequestHandler(string aMessageName,RequestHandler handler)
{
??? lock (this.m_EventHandlers)
??? {
???????? if (!this.m_EventHandlers.Contains(aMessageName))
???????? {
???????????? this.m_EventHandlers.Add(aMessageName,handler);
???????? }
??? }
}

protected RequestHandler FindRequestHandler(string aMessageName)
{
??? lock (this.m_EventHandlers)
??? {
???????? RequestHandler handler = (RequestHandler)m_EventHandlers[aMessageName];
???????? return handler;
??? }
}

?????? AddRequestHandler()用于添加委托對象與aMessageName的鍵值對，而FindRequestHandler()方法用于查找 該委托對象。而抽象方法AddRequestHandlers()則留給Service的子類實現(xiàn)，簡單的實現(xiàn)如MyService的 AddRequestHandlers()方法：

C#語言:? public class MyService:Service
{
???? public MyService(string aServiceName):base(aServiceName)
???? {}

???? protected override void AddRequestHandlers()
???? {
???????? this.AddRequestHandler("Test1",new RequestHandler(Test1));
???????? this.AddRequestHandler("Test2",new RequestHandler(Test2));
???? }

???? private void Test1(string aMessageName,IMessageItemSequence aMessageBody,ref? IMessageItemSequence aReplyMessageBody)
???? {
???????? Console.WriteLine("MessageName:{0}\n",aMessageName);
???????? Console.WriteLine("MessageBody:{0}\n",aMessageBody);
???????? aReplyMessageBody.SetValue("result","Success");
???? }

???? private void Test2(string aMessageName,IMessageItemSequence aMessageBody,ref?? IMessageItemSequence aReplyMessageBody)
???? {
???????? Console.WriteLine("Test2" + aMessageBody.ToString());
???? }
}

?????? Test1和Test2方法均為匹配RequestHandler委托簽名的方法，然后在AddRequestHandlers()方法中，通過調(diào)用 AddRequestHandler()方法將這些方法與MessageName對應(yīng)起來，添加到m_EventHandlers中。

?????? 需要注意的是，本文為了簡要的說明這種處理方式，所以簡化了Test1和Test2方法的實現(xiàn)。而在實際開發(fā)中，它們才是實現(xiàn)具體業(yè)務(wù)的重要方法。而利用這種方式，則解除了服務(wù)之間依賴的耦合度，我們隨時可以為服務(wù)添加新的業(yè)務(wù)邏輯，也可以方便的增加服務(wù)。

?????? 通過這樣的設(shè)計，Service的OnRequest()方法的最終實現(xiàn)如下所示：

C#語言:? public IMessage OnRequest(IMessage aMessage)
{
??? string messageName = aMessage.GetMessageName();
??? string messageID = aMessage.GetMessageID();
??? IMessage message = m_Factory.CreateMessage();

??? IMessageItemSequence replyMessage = m_Factory.CreateMessageItemSequence();
??? RequestHandler handler = FindRequestHandler(messageName);
??? handler(messageName,aMessage.GetMessageBody(),ref replyMessage);

??? message.SetMessageName(messageName);
??? message.SetMessageID(messageID);
??? message.SetMessageBody(replyMessage);

??? return message;
}

?????? 利用這種方式，我們可以非常方便的實現(xiàn)服務(wù)間通信，以及客戶端與服務(wù)端間的通信。例如，我們分別在服務(wù)端定義MyService(如前所示)和TestService：

C#語言: public class TestService:Service
{
???? public TestService(string aServiceName):base(aServiceName)
???? {}

???? protected override void AddRequestHandlers()
???? {
???????? this.AddRequestHandler("Test1",new RequestHandler(Test1));????????
???? }

???? private void Test1(string aMessageName,IMessageItemSequence aMessageBody,ref? IMessageItemSequence aReplyMessageBody)
???? {??????????
???????? aReplyMessageBody = SendRequest("MyService",aMessageName,aMessageBody);
???????? aReplyMessageBody.SetValue("result2","Success");
???? }

}

?????? 注意在TestService中的Test1方法，它并未直接處理消息aMessageBody，而是通過調(diào)用SendRequest()方法，將其傳遞到MyService中。

?????? 對于客戶端而言，情況比較特殊。根據(jù)前面的分析，我們知道除了發(fā)送消息的操作是在客戶端完成外，其他的具體執(zhí)行都在服務(wù)端實現(xiàn)。所以諸如 MyService和TestService等服務(wù)類，只需要部署在服務(wù)端即可。而客戶端則只需要定義一個實現(xiàn)Service的空類即可：

C#語言: public class MyServiceClient:Service
{
??? public MyServiceClient(string aServiceName):base(aServiceName)
???? {}

???? protected override void AddRequestHandlers()
???? {}
}

?????? MyServiceClient類即為客戶端定義的服務(wù)類，在AddRequestHandlers()方法中并不需要實現(xiàn)任何代碼。如果我們在 Service抽象類中，將AddRequestHandlers()方法定義為virtual而非abstract方法，則這段代碼在客戶端服務(wù)中也可 以省去。另外，客戶端服務(wù)類中的aServiceName可以任意賦值，它與服務(wù)端的服務(wù)名并無實際聯(lián)系。至于客戶端具體會調(diào)用哪個服務(wù)，則由 SendRequest()方法中的aServiceName決定：

C#語言: IMessageFactory factory = new MessageFactory();
IMessageItemSequence body = factory.CreateMessageItemSequence();
//……
MyServiceClient service = new MyServiceClient("Client");
IMessageItemSequence reply = service.SendRequest("TestService","Test1",body);

??????? 對于service.SendRequest()的執(zhí)行而言，會先調(diào)用TestService的Test1方法；然后再通過該方法向MyService發(fā)送，最終調(diào)用MyService的Test1方法。

?????? 我們還需要另外定義一個類，負(fù)責(zé)添加服務(wù)，并初始化這些服務(wù)：

C#語言:? public class Server
{
???? public Server()
???? {
???????? m_Services = new ArrayList();
???? }
???? private ArrayList m_Services;???
???? public void AddService(IListenService service)
???? {
???????? this.m_Services.Add(service);
???? }

???? public void Initialize()
???? {?

???????? IDictionary tcpProp = new Hashtable();
???????? tcpProp["name"] = "tcp9090";
???????? tcpProp["port"] = 9090;

???????? TcpChannel channel = new TcpChannel(tcpProp,new BinaryClientFormatterSinkProvider(),
???????????????????????????????????????????????????? new BinaryServerFormatterSinkProvider());??????????
???????? ChannelServices.RegisterChannel(channel);
???????? foreach (Service service in m_Services)
???????? {
????????????? service.Initialize();
???????? }??????????
???? }
}

?????? 同理，這里的Channel，IP和Port均應(yīng)通過配置文件讀取。最終的類圖如下所示：

?

?????? 在服務(wù)端，可以調(diào)用Server類來初始化這些服務(wù)：

C#語言:? static void Main(string[] args)
{?
??? MyService service = new MyService("MyService");
??? TestService service1 = new TestService("TestService");


??? Server server = new Server();
??? server.AddService(service);
??? server.AddService(service1);

??? server.Initialize();
??? Console.ReadLine();
}

?四、結(jié)論

?????? 利用這個基于消息與.Net Remoting技術(shù)的分布式架構(gòu)，可以將企業(yè)的業(yè)務(wù)邏輯轉(zhuǎn)換為對消息的定義和處理。要增加和修改業(yè)務(wù)，就體現(xiàn)在對消息的修改上。服務(wù)間的通信機(jī)制則完全 交給整個架構(gòu)來處理。如果我們將每一個委托所實現(xiàn)的業(yè)務(wù)（或者消息）理解為Contract，則該結(jié)構(gòu)已經(jīng)具備了SOA的雛形。當(dāng)然，該架構(gòu)僅僅處理了消 息的傳遞，而忽略了對底層事件的處理（類似于Corba的Event Service），這個功能我想留待后面實現(xiàn)。

?????? 唯一遺憾的是，我缺乏驗證這個架構(gòu)穩(wěn)定性和效率的環(huán)境。應(yīng)該說，這個架構(gòu)是我們在企業(yè)項目解決方案中的一個實踐。但是解決方案則是利用了CORBA中間 件，在Unix環(huán)境下實現(xiàn)并運行。本架構(gòu)僅僅是借鑒了核心的實現(xiàn)思想和設(shè)計理念，從而完成的在.Net平臺下的移植。由于Unix與Windows Server的區(qū)別，其實際的優(yōu)勢還有待驗證。

?

轉(zhuǎn)載于:https://blog.51cto.com/wayfarer/279909

總結(jié)

以上是生活随笔為你收集整理的基于消息与.Net Remoting的分布式处理架构的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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