?

本章內(nèi)容：
● 消息隊列概述
● 消息隊列體系結(jié)構(gòu)
● 消息隊列管理工具
● 編程實現(xiàn)消息隊列
● 課程訂單示例應(yīng)用程序
● 消息隊列和WCF

消息隊列的編程實現(xiàn)

既然理解了消息隊列的體系結(jié)構(gòu)之后，就可以探討其編程了。下面幾節(jié)將學習如何創(chuàng)建和控制 隊列，如何發(fā)送和接收消息。

還要構(gòu)建一個小型課程訂單應(yīng)用程序，它由發(fā)送部分和接收部分組成。

創(chuàng)建消息隊列

創(chuàng)建消息隊列 前面了解了如何使用Computer Management 實用程序創(chuàng)建消息隊列。消息隊列還可以用 MessageQueue 類的Create()方法以編程方式創(chuàng)建。
在 Create()方法中，必須傳遞新隊列的路徑。路徑包括隊列所在主機的名稱和隊列的名稱。在下 面的例子中，要在本地主機上創(chuàng)建MyNewPublicQueue 隊列。為了創(chuàng)建私有隊列，路徑名必須包含 Private，如\MyNewPrivateQueue。

調(diào)用 Create()方法之后，就可以修改隊列的屬性。例如，使用Label 屬性，把隊列的標簽設(shè)置為 Demo Queue。示例程序把隊列的路徑和格式名寫到控制臺上。格式名用UUID 自動創(chuàng)建，UUID 可 用于訪問隊列，且無須服務(wù)器名：

using System;
using System.Messaging;
namespace Wrox.ProCSharp.Messaging
{
class Program
{
    static void Main()
    {
        using (var queue = MessageQueue.Create(@".\MyNewPublicQueue"))
        {
            queue.Label = "Demo Queue";
            Console.WriteLine("Queue created:");
            Console.WriteLine("Path: {0}", queue.Path);
            Console.WriteLine("FormatName: {0}", queue.FormatName);
        }
    }
  }
}

創(chuàng)建隊列時需要管理權(quán)限。通常不希望應(yīng)用程序的用戶擁有管理權(quán)限。這就是隊 列通常用安裝程序創(chuàng)建原因.

查找隊列

路徑名和格式名可以用于標識隊列。要查找隊列，必須區(qū)分公共隊列和私有隊列。公共隊列在 Active Directory 中發(fā)布。對于這些隊列，無須知道它們所在的系統(tǒng)。只有在己知隊列所在系統(tǒng)私有 隊列名稱時才能找到私有隊列。
在 Active Directory 域中搜索隊列的標簽、類別或格式名，就可以找到公共隊列。還可以獲得計算 機上的所有隊列。MessageQueue 類的靜態(tài)方法GetPublicQueuesByLabel()、GetPublicQueuesByCategory() 和GetPublicQueuesByMachine()可以搜索隊列。GetPublicQueues()方法返回包含域中所有公共隊列的 數(shù)組。

using System;
using System.Messaging;
namespace Wrox.ProCSharp.Messaging
{
    class Program
    {
      static void Main()
      {
          foreach (var queue in MessageQueue.GetPublicQueues())
          {
              Console.WriteLine(queue.Path);
          }
      }
    }
}

GetPublicQueues()方法是重載的。它的一個重載版本允許傳遞MessageQueueCriteria 類的一個實 例。利用這個類可以搜索在某個時刻之前或之后創(chuàng)建或修改的隊列，還可以查找隊列的類別、標簽 或計算機名。

可以使用靜態(tài)方法GetPrivateQueuesByMachine()搜索私有隊列。這個方法返回指定系統(tǒng)中的所 有私有隊列。

打開已知隊列

如果隊列名已知，就不需要搜索它。使用路徑或格式名就可以打開隊列。路徑或格式名都在 MessageQueue 類的構(gòu)造函數(shù)中設(shè)置。

路徑名

路徑指定了打開隊列需要的計算機名和隊列名。下面的代碼示例打開本地主機上的 MyPublicQueue 隊列。為了確定隊列是否存在，可以使用靜態(tài)方法MessageQueue.Exists()：

using System;
using System.Messaging;
namespace Wrox.ProCSharp.Messaging
{
    class Program
    {
        static void Main()
        {
            if (MessageQueue.Exists(@".\mynewpublicqueue"))
            {
                var queue = new MessageQueue(@".\mynewpublicqueue");
                //...
            }
            else
            {
                Console.WriteLine("Queue .\MyPublicQueue not existing");
            }
        }
    }
}

根據(jù)隊列的類型，在打開隊列時需要不同的標識符。下表 列出了指定類型的隊列名的語法。

在使用路徑名打開公共隊列時，需要傳遞計算機名。如果計算機名未知，則可以使用格式名代 替。私有隊列的路徑名只能在本地系統(tǒng)上使用，必須使用格式名遠程訪問私有隊列。

格式名

除了路徑名之外，還可以使用格式名打開隊列。格式名用于在Active Directory 中搜索隊列，獲 得隊列所在的主機。在斷開連接的環(huán)境下，在發(fā)送消息時隊列不能到達，此時就需要使用格式名：

MessageQueue queue = new MessageQueue(
@"FormatName:PUBLIC=09816AFF-3608-4c5d-B892-69754BA151FF");

格式名還有一些其他用途。它可以用于打開私有隊列，并指定要使用的協(xié)議：
要 訪 問 私有隊列， 必須給構(gòu)造函數(shù)傳遞字符串 FormatName:PRIVATE=MachineGUID\QueueNumber。在創(chuàng)建隊列時，會生成私有隊列的隊 列號。隊列號在\System32\msmq\storage\lqs 目錄下。
使用 FormatName:DIRECT=Protocol:MachineAddress\QueueName，可以指定用于發(fā)送消息的 協(xié)議。Message Queuing 3.0 及以后版本支持HTTP 協(xié)議。

FormatName:DIRECT=OS:MachineName\QueueName 是使用格式名指定隊列的另一種方式。 此時不需要指定協(xié)議，但仍可以使用計算機名和格式名。

發(fā)送消息

可以使用MessageQueue 類的Send()方法給隊列發(fā)送消息。作為參數(shù)傳遞給Send()方法的對象序 列化到相關(guān)聯(lián)的隊列上。Send()方法是重載的，這樣才能傳遞標簽和MessageQueueTransaction 對象。 Message Queuing 的事務(wù)行為在后面論述。 下面的代碼示例先檢查隊列是否存在，如果不存在，就創(chuàng)建一個隊列。接著打開隊列，使用Send() 方法給隊列發(fā)送Sample Message 消息。 路徑名給服務(wù)器名指定“.”，表示它是本地系統(tǒng)。私有隊列的路徑名只能在本地使用。

using System;
using System.Messaging;
namespace Wrox.ProCSharp.Messaging
{
    class Program
    {
        static void Main()
        {
            try
            {
                if (!MessageQueue.Exists(@".\Private$\MyPrivateQueue"))
                {
                    MessageQueue.Create(@".\Private$\MyPrivateQueue");
                }
                var queue = new MessageQueue(@".\Private$\MyPrivateQueue");
                queue.Send("Sample Message", "Label");
          }
          catch (MessageQueueException ex)
          {
              Console.WriteLine(ex.Message);
          }
      }
  }
}

消息格式化程序

消息傳輸給隊列的格式取決于格式化程序。MessageQueue 類有一個Formatter 屬性，通過它可 以指定格式化程序。默認的格式化程序XmlMessageFormatter 會用XML 語法格式化消息，如前面的 例子所示。 消息格式化程序?qū)崿F(xiàn)IMessageFormatter 接口。System.Messaging 名稱空間中有3 個消息格式化 程序： XmlMessageFormatter 是默認的格式化程序，它使用XML 序列化對象，XML 格式的內(nèi)容詳。

使用 BinaryMessageFormatter，可以用二進制格式對消息進行序列化。這些消息比使用XML 格式化的消息短。

ActiveXMessageFormatter 是一個二進制格式化程序，這樣可以用COM 對象讀寫消息。使 用這個格式化程序，可以用.NET 類把消息寫入隊列中，使用COM 對象從隊列中讀取消息， 反之亦然。

發(fā)送復(fù)雜的消息

除了傳遞字符串之外，還可以給MessageQueue 類的Send()方法傳遞對象。雖然該類的類型必 須滿足一些特定的要求，但它們?nèi)Q于格式化程序。

對于二進制格式化程序，該類必須用[Serializable]屬性序列化。使用.NET 運行庫的序列化功能， 序列化所有字段(包括私有字段)。實現(xiàn)ISerializable 接口就可以定義自定義序列化。

XML 序列化在使用XML 格式化程序時進行。在XML 序列化過程中，會序列化所有公共字段 和屬性。使用System.Xml.Serialization 名稱空間中的屬性可以影響XML 序列化。

接收消息

要讀取消息，也可以使用MessageQueue 類。通過Receive()方法可以讀取一條消息，再將該消 息從隊列中刪除。如果使用不同的優(yōu)先級發(fā)送消息，就讀取優(yōu)先級最高的消息。讀取優(yōu)先級相同的 消息時，第一條發(fā)送的消息不一定是第一條讀取的消息，因為消息在網(wǎng)絡(luò)中的傳遞順序無法保證。 要保證發(fā)送順序和讀取順序相同，可以使用事務(wù)消息隊列。

在下面的例子中，要從私有隊列MyPrivateQueue 中讀取一條消息。之前把一個簡單的字符串傳 遞給該消息。在使用XmlMessageFormatter 格式化程序讀取消息時，必須把要讀取的對象的類型傳 遞給該格式化程序的構(gòu)造函數(shù)。在本例中，將System.String 類型傳遞給XmlMessageFormatter 的構(gòu) 造函數(shù)的參數(shù)數(shù)組。這個構(gòu)造函數(shù)可以接收一個String 數(shù)組，該數(shù)組包含要作為字符串傳遞的類型； 也可以接收一個Type 數(shù)組。

用 Receive()方法讀取消息，再把消息正文寫入控制臺中：

using System;
using System.Messaging;
namespace Wrox.ProCSharp.Messaging
{
    class Program
    {
        static void Main()
        {
            var queue = new MessageQueue(@".\Private$\MyPrivateQueue");
            queue.Formatter = new XmlMessageFormatter(
new string[] {"System.String"});
            Message message = queue.Receive();
            Console.WriteLine(message.Body);
      }
  }
}

Receive()方法將同步執(zhí)行，如果隊列中沒有消息，它就會等待隊列中有消息時再執(zhí)行。

枚舉消息

除了使用 Receive()方法逐條消息地讀取之外，還可以使用枚舉器遍歷所有消息。因為 MessageQueue 類實現(xiàn)IEnumerable 接口，所以可以在foreach 語句中使用。使用迭代器時，雖然消息 不會從隊列中刪除，但可以查看消息從而獲得它們的內(nèi)容：

var queue = new MessageQueue(@".\Private$\MyPrivateQueue");
queue.Formatter = new XmlMessageFormatter(
new string[] {"System.String"});
foreach (Message message in queue)
{
    Console.WriteLine(message.Body);
}

除了使用IEnumerable 接口外，還可以使用MessageEnumerator 類。雖然MessageEnumerator 類實現(xiàn) IEnumerator 接口，但它有更多功能。實現(xiàn)IEnumerable 接口，就表示不從隊列中刪除消息。 MessageEnumerator 類的RemoveCurrent()方法可以從枚舉器的當前光標位置刪除消息。

在下面的例子中，使用MessageQueue 類的GetMessageEnumerator()方法訪問MessageEnumerator 類。通過MessageEnumerator 類的MoveNext()方法，可以逐條查看消息。MoveNext()方法重載為允 許把一個時間段作為參數(shù)。這是使用這個枚舉器的一個主要優(yōu)點。現(xiàn)在，線程可以在指定的時間段 內(nèi)等待消息到達隊列，之后就不等待了。IEnumerator 接口定義的Current 屬性返回消息的一個引用：

var queue = new MessageQueue(@".\Private$\MyPrivateQueue");
queue.Formatter = new XmlMessageFormatter(
new string[] {"System.String"});
using (MessageEnumerator messages = queue.GetMessageEnumerator())
{
    while (messages.MoveNext(TimeSpan.FromMinutes(30)))
    {
        Message message = messages.Current;
        Console.WriteLine(message.Body);
    }
}

異步讀取

MessageQueue 類的Receive()方法會等到隊列中的消息可以讀取為止。為了避免阻礙線程的執(zhí) 行，可以在Receive()方法的一個重載版本中指定一個超時期限。要在超時后讀取隊列中的消息，必 須再次調(diào)用Receive()方法。除了輪詢消息外，還可以調(diào)用BeginReceive()異步方法。在使用 BeginReceive()開始異步讀取消息之前，應(yīng)設(shè)置ReceiveCompleted 事件。ReceiveCompleted 事件需要 ReceiveCompletedEventHandler 委托，在消息到達隊列并可以讀取時該委托引用要調(diào)用的方法。在下 面的例子中，把MessageArrived()方法傳遞給ReceivedCompletedEventHandler 委托：

var queue = new MessageQueue(@".\Private$\MyPrivateQueue");
queue.Formatter = new XmlMessageFormatter(
new string[] { "System.String" });
queue.ReceiveCompleted += MessageArrived;
queue.BeginReceive();
// thread does not wait

MessageArrived()處理程序方法需要兩個參數(shù)。第一個參數(shù)是MessageQueue 事件源。第二個參 數(shù)是ReceiveCompletedEventArgs 類型，它包含消息和異步結(jié)果。在下面的例子中，調(diào)用隊列中的 EndReceive()方法，以獲得異步方法的結(jié)果，即消息：

public static void MessageArrived(object source, ReceiveCompletedEventArgs e)
{
    MessageQueue queue = (MessageQueue)source;
    Message message = queue.EndReceive(e.AsyncResult);
    Console.WriteLine(message.Body);
}

閱讀原文：原文鏈接