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

C# и .NET — Сетевое программирование — Программирование с использованием Message Queuing

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

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

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

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

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

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

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

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

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

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

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

Путевое имя

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

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

Синтаксис идентификаторов для очередей разных типов

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

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

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

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

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

Для доступа к частной очереди строка, передаваемая конструктору, выглядит так: FormatName:PRIVATE=MachineGUID\QueueNumber . Номер QueueNumber для частных очередей генерируется при их создании. Номера очередей можно просмотреть в каталоге \System32\msmq\storage\lqs.

В FormatName:DIRECT=Protocol:MachineAddress\QueueName можно указать протокол, который должен использоваться для отправки сообщений. Протокол HTTP поддерживается, начиная с версии Message Queuing 3.0.

FormatName:DIRECT=OS:MachineName\QueueName — другой способ задания очереди с применением форматного имени. Таким образом, протокол указывать не нужно, а только имя машины и форматное имя.

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

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

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

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

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

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

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

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

XmlMessageFormatter

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

BinaryMessageFormatter

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

ActiveXMessageFormatter

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

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

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

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

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

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

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

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

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

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

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

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

Обзор Message Queuing

В пространстве имен System.Messaging определены классы, которые позволяют выполнять чтение и запись сообщений с использованием такого предлагаемого в составе операционной системы Windows средства для организации сообщений, как .

Обмен сообщениями может применяться в сценарии автономной работы, в котором не требуется, чтобы клиент и сервер обязательно запускались в одно и то же время.

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

При синхронном программировании, когда вызывается метод, то вызвавший его код должен ожидать, пока метод не завершит свою работу. При асинхронном программировании вызывающий поток запускает метод и параллельно продолжает свою работу. Асинхронное программирование основано на применении делегатов, библиотек классов, которые уже поддерживают асинхронные методы (например, прокси-классы веб-служб и классы из пространств System.Net и System.IO), либо специальных потоков. Как при синхронном, так и при асинхронном программировании клиент и сервер должны работать в одно и то же время.

Хотя Message Queuing работает асинхронно, поскольку клиент (отправитель) не ожидает прочтения получателем (сервером) оправленных ему данных, между Message Queuing и асинхронным программированием существует принципиальная разница: Message Queuing может выполняться в автономной (отключенной) среде. На момент отправки данных их получатель может быть отключен. Позднее, когда получатель подключится, он получит данные без вмешательства отправляющего приложения.

Подключенное и отключенное программирование можно сравнить с разговором по телефону и отправкой почтовых сообщений. При разговоре с кем-либо по телефону оба участника должны быть подключены одновременно — это синхронная коммуникация. В случае обмена электронной почтой отправитель не знает, когда его послание будет прочитано. Люди, использующие эту технологию, работают в автономном (отключенном) режиме.

Конечно, может случиться так, что почта не будет прочитана никогда, а просто проигнорирована. Такова природа отключенных коммуникаций. Чтобы избежать этой проблемы, можно запросить ответ или подтверждение факта прочтения письма. Если ответ не придет в течение определенного времени, возможно, придется как-то справляться с таким «исключением». Все это также возможно в Message Queuing.

Message Queuing, по сути, можно считать технологией для обмена электронными сообщениями между приложениями, а не людьми. Она обладает множеством функциональных возможностей, которые в других службах обмена сообщениями не доступны: гарантированием доставки, применением транзакций, получением подтверждений, экспресс-режимом, использующим память, и т.д. Message Queuing предлагает массу полезных средств для коммуникаций между приложениями.

С помощью Message Queuing можно отправлять, принимать и маршрутизировать сообщения в подключенной и отключенной (автономной) среде. На рисунке ниже показан очень простой способ использования сообщений. Отправитель посылает сообщения в очередь сообщений, а получатель принимает их из этой очереди:

Когда следует использовать Message Queuing

Одной из ситуаций, в которых удобно применять Message Queuing — это когда клиентское приложение часто отключается от сети (например, у коммивояжера, навещающего заказчиков на местах). Коммивояжер может вводить данные заказа непосредственно у заказчика. Приложение ставит сообщение о каждом заказе в очередь сообщений, находящуюся на клиентской системе. Как только коммивояжер возвращается в свой офис, заказ автоматически передается из очереди сообщений клиентской системы в очередь сообщений целевой системы, где и обрабатывается.

Помимо портативного компьютера, коммивояжер может использовать устройство Pocket Windows, где также доступно Message Queuing.

Технология Message Queuing может быть полезна и в подключенной среде. Представьте сайт электронной коммерции (показан на рисунке ниже), где в определенные периоды времени сервер полностью загружен обработкой заказов, например, в ранний вечер и в выходные, при этом по ночам нагрузка значительно уменьшается. Решение проблемы может состоять в приобретении более быстрого сервера или в добавлении дополнительных серверов к системе, чтобы они справлялись с пиковыми нагрузками.

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

Функциональные возможности Message Queuing

Технология Message Queuing является службой, которая поставляется как часть операционной системы Windows. Ниже перечислены ее основные функциональные возможности:

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

В экспресс-режиме сообщения могут пересылаться очень быстро. В экспресс-режиме сообщения просто сохраняются в памяти.

Для механизма восстановления сообщения могут отправляться с гарантированной доставкой. Такие сообщения сохраняются в файлах и доставляются даже в случае перезагрузки сервера.

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

В Message Queuing 3.0 поддерживается возможность отправки многоадресных (multicast) сообщений.

В Message Queuing 4.0 поддерживается возможность распознавания вредоносных сообщений. Для таких сообщений может быть определена специальная очередь.

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

В Message Queuing 5.0 поддерживаются более безопасные алгоритмы аутентификации, и может обрабатываться большее количество очередей. (В Message Queuing 4.0 при обработке нескольких тысяч очередей начинали возникать проблемы с производительностью.)

Из-за того, что Message Queuing является частью операционной системы, установить версию Message Queuing 5.0 в системе Windows XP или Windows Server 2003 не получится. Эта версия входит в состав ОС Windows Server 2008 R2 и Windows 7.

Продукты Message Queuing

Версия Message Queuing 5.0 поставляется в составе Windows 7 и Windows Server 2008 R2. В Windows 2000 входила версия Message Queuing 2.0, в которой не было поддержки ни протокола HTTP, ни многоадресных сообщений. Версия Message Queuing 3.0 поставлялась в составе Windows XP и Windows Server 2003, а версия Message Queuing 4.0 — в составе Windows Vista и Windows Server 2003.

При использовании ссылки Turn Windows Features on or off (Включение или отключение компонентов Windows), которая предлагается в Windows 7 в окне Configuring Programs and Features (Программы и компоненты), можно обнаружить отдельный раздел с опциями, касающимися Message Queuing:

В этом разделе доступны для выбора перечисленные ниже компоненты:

Microsoft Message Queue (MSMQ) Server Core

Основные компоненты сервера очереди сообщений (MSMQ), которые необходимы для получения базовой функциональности Message Queuing.

Active Directory Domain Services Integration

Интеграция MSMQ доменных служб Active Directory. Это средство позволяет записывать имена очередей сообщений в Active Directory. С помощью этой опции можно находить очереди в Active Directory и защищать их на основе пользователей и групп пользователей Windows.

MSMQ HTTP Support

Поддержка протокола HTTP MSMQ. Поддержка MSMQ HTTP позволяет отправлять и принимать сообщения, используя протокол HTTP.

Triggers

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

Multicast Support

Поддержка многоадресной рассылки. Позволяет отправлять сообщения группам серверов.

MSMQ DCOM Proxy

С помощью DCOM-прокси система может подключаться к удаленному серверу, используя API-интерфейс DCOM.

После установки Message Queuing в системе должна быть обязательно запущена служба Message Queuing (показана на рисунке). Эта служба читает и записывает сообщения, а также взаимодействует с другими серверами Message Queuing для осуществления маршрутизации сообщений по сети: