[MessageQueues] MSMQ

Mechanizm ten zintegrowany jest z systemami z rodziny Windows i Windows Server. Kolejkowanie włączamy jako feature Windowsa. Przykładowo na Win 10:

Kolejki możemy zakładać konfigurować i podglądać po przejściu do ekranu zarządzana komputerem (PPM na Komputer i wybór Zarządzaj).

Biblioteka kliencka wbudowana w jest .NET Framework i przestrzen nazw System.Messaging, Kolejki pracują w sposób synchroniczny (FIFO). Na każdej kolejce można ustawić transakcyjność oraz persystencję (MSMQ może zapisywać wiadomości na dysku)

Kolejki mogą być prywatne lub publiczne. Główna różnica polega na tym, że w przypadku publicznych kolejkek, możemy je zintegrować z Active Directory i mechanizmami security Windowsa.

Kolejki prywatne mogą być adresowane na dwa sposoby:
PATH:
- {machine}\private$\{queueName}, np:  .\private$\myqueue

DIRECT:
- DIRECT={protocol}:{address}\private$\{queueName}, np: DIRECT=TCP:192.168.2.140\private$\myqueue

Po założeniu kolejki możemy do niej wstawiać elementy w następujący sposób:

[HttpPost]
public ActionResult SendMessage(MessageEntity entity)
{
    string address = entity.UseTransaction ? ".\\private$\\transactionalcontactmessages" : ".\\private$\\contactmessages";
    using (var queue = new MessageQueue(address))
    {
        var message = new Message(new ContactMessage() { Content = entity.Content });
        message.Recoverable = entity.StoreOnDisk;

        if (entity.UseTransaction)
        {
            using (var transaction = new MessageQueueTransaction())
            {
                transaction.Begin();
                queue.Send(message, transaction);
                queue.Send(new Message("anotherMsg"), transaction);
                transaction.Commit();
            }
        }
        else
        {
            queue.Send(message);
        }
    }
    return new RedirectResult("/");
}

Transakcyjność ustawiamy w momencie zakładania kolejki. Wiadomości serializowane są do XML-a jako typ danych lub obiekt. Można wstawiać je także w dowolnym innym formacie, np. JSON ustawiając Stream na propercji Body. Wiadomości bez właściwości Recoverable zapisywane będą około 10 razy szybciej. Transakcja jest kolejnym narzutem czasowym przy zapisie, ale można dzięki niej wstawiać jednokrotnie wiele wiadomości.

Najprostszy kod przetwarzający może wyglądać w następujący sposób:

using (var queue = new MessageQueue(".\\private$\\contactmessages"))
{
    while (true)
    {
        var message = queue.Receive();
        Console.WriteLine("Processing");
    }
}

Podobnie można objąć całą operację w MessageQueueTransaction.

Request - Response

Zgodnie z tym wzorcem tworzone są dwie kolejki. Jedna strona komunikacji wstawia do pierwszej kolejki wiadomości, ustawiając property ResponseQueue. Dzięki temu, druga strona po przetworzeniu wiadomości wstawia odpowiedź do podanej kolejki. Dla każdej wiadomości można tworzyć nową kolejkę odpowiedzi.

private void RequestResponse(string queueAddress, string content)
{
    var responseQueue = MessageQueue.Create(GetPrivateQueueAddress());
    using (var queue = new MessageQueue(queueAddress))
    {
        var message = new Message(new ContactMessage() { Content = content });
        message.ResponseQueue = responseQueue;
        queue.Send(message);

        //Waiting for message
        var response = responseQueue.Receive();
    }
}

private string GetPrivateQueueAddress()
{
    return string.Format(".\\private$\\{0}", Guid.NewGuid().ToString().Substring(0, 6));
}

Kod po drugiej stronie może wyglądać w następujący sposób:

while (true)
{
    var message = queue.Receive();
    if(message.ResponseQueue != null)
    {
        using(var responseQueue = message.ResponseQueue)
        {
            var msg = new Message();
            msg.Label = "Processed";
            responseQueue.Send(msg);
        }                        
    }

    Console.WriteLine("Processing");
}

Kolejki mogą także działać w modelu Publish-Subscribe. Ich adres ustawiamy wtedy na wirtualny IP, na którym nasłuchują odbiorcy. Przykładowy adres takiej kolejki to: "FormatName:MULTICAST=234.1.1.2:8001". Wszyscy subskrybenci muszą mieć ustawione taką wartość w property MulticastAddress.