Программирование с использованием Message Queuing

122

Теперь, когда архитектура Message Queuing известна, можно приступать к программированию. В последующих разделах будет показано, как создавать и управлять очередями, а также как отправлять и принимать сообщения.

Создание очереди сообщений

Вы уже видели, как создаются очереди сообщений утилитой Computer Management. Но очереди сообщений могут быть созданы и программно вызовом метода Create() класса MessageQueue. Методу Create() должен быть передан путь к новой очереди. Этот путь состоит из имени хоста, где расположена очередь, и имени очереди.

В следующем примере очередь MyNewPrivateQueue создается на локальном хосте. Чтобы создать частную очередь, путь должен включать Private$; например, \Private$\MyNewPrivateQueue.

После вызова метода Create() свойства очереди могут быть изменены. Например, используя свойство Label, установим метку очереди в Demo Queue. В примере программы путь очереди и форматное имя выводятся на консоль. Форматное имя создается автоматически с UUID, который может применяться для доступа к очереди без указания имени сервера:

using System;
using System.Messaging;

namespace MSMQSample
{
    class Program
    {
        static void Main(string[] args)
        {
            using (var queue = MessageQueue.Create(".\\private$\\MyNewPrivateQueue"))
            {
                queue.Label = "Demo Queue";
                Console.WriteLine("Очередь создана:");
                Console.WriteLine("Путь: {0}", queue.Path);
                Console.WriteLine("Форматное имя: {0}", queue.FormatName);
            }
            Console.ReadLine();
        }
    }
}
Создание новой очереди сообщений

Для создания очереди необходимы административные привилегии. Обычно нельзя рассчитывать, что пользователь приложения будет иметь их. Именно поэтому очереди обычно создаются программой установки.

Нахождение очереди

Для идентификации очередей могут использоваться путевое имя и форматное имя. При поиске очереди необходимо делать различия между общедоступными и частными очередями. Общедоступные очереди публикуются в Active Directory. Для таких очередей не обязательно знать систему, на которой они расположены. Частные очереди могут быть найдены только в случае, если известно имя системы, на которой расположена очередь:

using System;
using System.Messaging;

namespace MSMQSample
{
    class Program
    {
        static void Main(string[] args)
        {
            foreach (var queue in MessageQueue.GetPrivateQueuesByMachine(Environment.MachineName))
                Console.WriteLine("Очередь: {0}\n", queue.Path);

            Console.ReadLine();
        }
    }
}
Поиск очередей

Общедоступные очереди в домене Active Directory можно искать по метке очереди, категории или форматного имени. Можно также получить все очереди, имеющиеся на машине. В классе MessageQueue предусмотрены статические методы для поиска: GetPublicQueuesByLabel(), GetPublicQueuesByCategory() и GetPublicQueuesByMachine(). Метод GetPublicQueues() возвращает массив всех общедоступных очередей в домене.

Открытие известных очередей

Если имя очереди известно, то искать ее не обязательно. Очереди могут открываться с использованием пути или форматного имени. И то, и другое может быть установлено конструктором класса MessageQueue.

Путевое имя

Путь указывает имя машины и имя очереди для ее открытия. В следующем примере кода открывается очередь MyNewPrivateQueue на локальном хосте. Чтобы удостовериться в существовании очереди, применяется статический метод MessageQueue.Exists():

if (MessageQueue.Exists(@".\private$\MyNewPrivateQueue"))
{
       var queue = new MessageQueue(@".\private$\MyNewPrivateQueue");
}
else
{
       Console.WriteLine("Очередь не существует");
}

В зависимости от типа очереди при открытии очередей должны использоваться разные идентификаторы. В следующей таблице показан синтаксис идентификаторов для очередей разных типов:

Синтаксис идентификаторов для очередей разных типов
Тип очереди Синтаксис
Общедоступная очередь ИмяКомпьютера\ИмяОчереди
Частная очередь ИмяКомпьютера\Private$\ИмяОчереди
Журнальная очередь ИмяКомпьютера\ИмяОчереди\Journal$
Журнальная очередь машины ИмяКомпьютера\Journal$
Очередь “мертвых писем” машины ИмяКомпьютера\DeadLetter$
Транзакционная очередь “мертвых писем” машины ИмяКомпьютера\XactDeadLetter$

Если для открытия общедоступной очереди используется путевое имя, необходимо передавать имя машины. Если имя машины не известно, вместо него может быть указано форматное имя. Путевое имя частной очереди может применяться только на локальных системах. Форматное имя должно использоваться для удаленного доступа к частным очередям.

Форматное имя

Вместо использования путевого имени для открытия очереди можно применять форматное имя. Форматное имя применяется для поиска очереди в Active Directory, чтобы получить хост, на котором расположена очередь. В автономной среде, где очередь во время отправки сообщения недоступна, необходимо использовать форматное имя:

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

Форматное имя используется по-разному. С его помощью можно открывать частные очереди и указывать нужный протокол.

Отправка сообщения

Для отправки сообщения в очередь используется метод Send() класса MessageQueue. Объект, переданный в качестве аргумента методу Send(), сериализуется в ассоциированную очередь. Метод Send() перегружен так, что можно передавать метку и объект MessageQueueTransaction.

В приведенном ниже примере кода сначала выполняется проверка, существует ли очередь. Если очередь не существует, она создается. Затем очередь открывается и с помощью метода Send() в очередь отправляется сообщение “Sample Message”. Путевое имя специфицирует точку вместо имени сервера, что означает локальную систему. Путевые имена частных очередей работают только локально:

try
{
      if (!MessageQueue.Exists(@".\private$\MyNewPrivateQueue"))
      {
              MessageQueue.Create(@".\private$\MyNewPrivateQueue");
      }

      var queue = new MessageQueue(@".\private$\MyNewPrivateQueue");
      queue.Send("Sample Message", "Label");
}
catch (MessageQueueException ex)
{
      Console.WriteLine(ex.Message);
}

На рисунке ниже показано окно оснастки Computer Management, в котором можно просматривать сообщения, появляющиеся в очереди:

Просмотр сообщений в очереди

Открыв сообщение и выбрав в диалоговом окне вкладку Body (Тело), можно увидеть, что сообщение сформатировано с использованием XML. Способ форматирования сообщения — функция форматировщика, ассоциированного с очередью сообщений.

Просмотр тела сообщения в формате XML

Форматировщик сообщений

Формат, в котором передаются сообщения в очередь, зависит от используемого форматировщика. Класс MessageQueue имеет свойство Formatter, через которое очереди может быть назначен объект-форматировщик.

Форматировщик по умолчанию — XmlMessageFormatter — форматирует сообщение в синтаксисе XML, как показано в предыдущем примере. Форматировщик сообщений реализует интерфейс IMessageFormatter. В пространстве имен System.Messaging доступны три форматировщика сообщений:

XmlMessageFormatter

Форматировщик по умолчанию. Он сериализует объекты, используя XML.

BinaryMessageFormatter

С помощью форматировщика BinaryMessageFormatter сообщения сериализуются в двоичный формат. Эти сообщения короче, чем сформатированные с применением XML.

ActiveXMessageFormatter

Двоичный форматировщик, так что сообщения могут быть прочитаны и записаны объектами COM. Используя этот форматировщик, можно записывать сообщения в очередь с помощью классов .NET и читать их оттуда объектами COM, и наоборот.

Простое сообщение, показанное на выше в формате XML, на рисунке ниже сформатировано с помощью BinaryMessageFormatter:

Просмотр сообщения, сформатированного BinaryMessageFormatter

Отправка сложных сообщений

Вместо строк методу Send() класса MessageQueue можно передавать объекты. Тип класса таких объектов должен соответствовать определенным требованиям, но они зависят от форматировщика.

Для двоичного форматировщика класс должен быть сериализуемым и снабжен атрибутом [Serializable]. При сериализации исполняющей системой .NET сериализуются все поля (включая приватные). Специальная сериализация может быть определена за счет реализации интерфейса ISerializable.

Сериализация XML происходит с помощью XML-форматировщика. При сериализации XML сериализуются все общедоступные поля и свойства. На сериализацию XML могут оказывать влияние атрибуты из пространства имен System.Xml.Serialization.

Прием сообщений

Для чтения сообщений можно использовать класс MessageQueue. Метод Receive() читает единственное сообщение и удаляет его из очереди. Если сообщения отправлены с разными приоритетами, читается сообщение с наивысшим приоритетом. Чтение сообщений с одинаковым приоритетом не обеспечивает поступление сообщений в порядке их отправки, потому что порядок сообщений в сети не гарантируется. Для получения гарантированного порядка применяйте транзакционные очереди сообщений.

В следующем примере сообщение читается из частной очереди MyNewPrivateQueue. Ранее в сообщение была передана простая строка. При чтении сообщения с использованием XmlMessageFormatter необходимо передавать типы прочитанных объектов конструктору форматировщика. В данном примере тип System.String передается в массив аргументов конструктора XmlMessageFormatter. Этот конструктор принимает либо массив String с типами в виде строк, либо массив объектов Type. Сообщение читается методом Receive() и затем тело сообщения выводится на консоль:

var queue = new MessageQueue(@".\private$\MyNewPrivateQueue");
queue.Formatter = new XmlMessageFormatter(
      new String[] { "System.String" });
Message message = queue.Receive();
Console.WriteLine(message.Body);

Метод Receive() ведет себя синхронно и ожидает появления сообщения в очереди, если на момент его вызова там было пусто.

Перечисление сообщений

Вместо чтения сообщений друг за другом с помощью метода Receive() можно применить перечислитель для прохождения сразу по всем сообщениям в очереди. Класс MessageQueue реализует интерфейс IEnumerable, и потому может быть использован в операторе foreach или LINQ. Здесь сообщения не удаляются из очереди, вы только заглядываете в каждое сообщение для извлечения содержимого:

var queue = new MessageQueue(@".\private$\MyNewPrivateQueue");
      queue.Formatter = new BinaryMessageFormatter();

foreach (Message message in queue)
      Console.WriteLine(message.Body);
Пройди тесты
Лучший чат для C# программистов