MSMQ: Pub/Sub

Esse post é a continuação de uma pequena série de posts sobre padrões básicos de integração usando o MSMQ com C#. Dessa vez, vou demonstrar como podemos implementar o padrão Pub/Sub, que também é conhecido como Publish/Subscribe. Esse padrão é mais ou menos uma versão entre sistemas do padrão Observer que encontramos na orientação a objetos. Pub/Sub é o padrão fundamental de uma Arquitetura Orientada a Eventos.

O Publisher notifica sobre a ocorrência de um evento sem saber quem ou até mesmo se existe algum Subscriber (também conhecido como inscrito, interessado, subscrito, etc. [nunca achei uma boa tradução para esse termo]) que irá recebê-lo.

Uma analogia a esse padrão seria a transmissão de um programa de rádio, por exemplo. A antena de transmissão simplesmente irradia o sinal sem saber quem irá sintonizá-lo -- esse seria o Publisher. Os subscribers, seriam os ouvintes.

Chega de teoria e vamos à prática!

Pré-Requisitos

Se você leu algum dos posts anteriores, sabe quais são os pré-requisitos. Caso contrário, dê uma olhada aqui na seção Pré-Requisitos.

Prática

O primeiro passo é criarmos um modelo canônico que será referenciado pelos participantes da solução. Sendo assim, crie um projeto do tipo class library em C# de nome Eip.Messages. Declare a seguinte interface seguida de uma implementação padrão.

public interface IEventOccurred
{
 string Event { get; set; }
}

[Serializable]
public class EventOccurredImpl
 : IEventOccurred
{
 public EventOccurredImpl(string @event)
 {
  Event = @event;
 }
 public string Event { get; set; }
}

Agora, adicione um novo projeto à solução do tipo Console de nome Eip.Publisher e adicione referências a System.Messaging e ao projeto que acabamos de criar. Em seguida, cole o trecho de código abaixo dentro do método static void Main(string[] args).

Console.Title = "Publisher Endpoint";
Console.WindowHeight = 10;
Console.WindowWidth = 50;

// Instancia a fila da publicação/notificação (publish)
MessageQueue publishQueue = new MessageQueue("FormatName:MULTICAST=236.1.1.1:8001");

while (true)
{
 Console.WriteLine("Entre com a descrição do evento ocorrido:");
 string inputEvent = Console.ReadLine();

 if (inputEvent == "sair")
  break;

 // Instancia a notificação
 IEventOccurred request = new EventOccurredImpl(inputEvent);

 // Instancia a mensagem que irá encapsular nossa notificação
 Message publishMessage = new Message(request, new BinaryMessageFormatter());

 // Envia a requisição
 publishQueue.Send(publishMessage);
 
 Console.WriteLine("Evento publicado: '{0}'", inputEvent);
}

Inicie a aplicação para ver se tudo está correto. Se tiver, vamos para o próximo passo,

Agora, adicione mais um projeto do tipo Console à solução com nome Eip.Subscriber. Aqui também, adicione referências a System.Messaging e a Eip.Messages. Em seguida, cole o trecho de código abaixo dentro do método static void Main(string[] args).

Console.Title = "Subscriber Endpoint";
Console.WindowHeight = 10;
Console.WindowWidth = 50;

// Cria uma fila exclusiva privada para esse subscriber
Console.WriteLine("Entre com o nome desse subscriber (apenas letras e underscore):");
string subscriberQueueName = Console.ReadLine();

string subscriberQueuePath = string.Format(@".\private$\{0}", subscriberQueueName);

MessageQueue subscribeQueue = null;

// Verifica se a fila já existe
// Se existir, instancia a fila; caso contrário, cria uma nova
if (MessageQueue.Exists(subscriberQueuePath))
 subscribeQueue = new MessageQueue(subscriberQueuePath);
else
 subscribeQueue = MessageQueue.Create(subscriberQueuePath);

// Note que estamos usando o mesmo IP declarado no Publisher
// Fazendo isso, estamos dizendo que a nossa fila irá receber as notificações Multicast
// recebidas nesse endereço.
subscribeQueue.MulticastAddress = "236.1.1.1:8001";

subscribeQueue.Formatter = new BinaryMessageFormatter();

Console.WriteLine("Aguardando notificações...");

// Loop principal para ouvir as notificações
while (true)
{
 Message notificationMessage = subscribeQueue.Receive();
 IEventOccurred eventReceived = (IEventOccurred)notificationMessage.Body;
 Console.WriteLine("Notificação recebida: '{0}'", eventReceived.Event);
}

Inicie essa aplicação também para ver se tudo está funcionando. Se estiver, você deverá ver algo como à seguir.

Para que esse tutorial faça sentido, inicie uma instância de Eip.Publisher e duas ou mais instâncias de Eip.Subscriber (entrando com nomes diferentes). O vídeo à seguir mostra como iniciar um Publisher e três Subscribers.

Conclusão

Com esse post, encerro a série de padrões elementares de uma arquitetura usando um sistema de mensagens como infraestrutura -- MSMQ. O padrão Pub/Sub em especial, é a alma de uma Arquitetura Orientada a Eventos. Vale ressaltar que essa abordagem, de forma alguma, visa substituir uma SOA; é um complemento. Mesmo porque SOA não é uma técnica, é um conceito -- você pode muito bem disponibilizar serviços em formato de mensagens.

Como já mencionado no primeiro post, foi proposital o nível de hard code e falta de abstração. Da mesma forma que você usaria uma interface de acesso a dados em uma camada de negócio, você também usaria uma abstração para o acesso à infraestrutura do sistema de mensageria em um cenário real.

Se você se interessou pelo assunto, eu recomendo dar uma olhada no sistema de mensagens RabbitMQ. O RabbitMQ é muito superior ao MSMQ em muitas questões. O MSMQ tem basicamente suporte a enviar e receber mensagens, enquanto que o RabbitMQ possui um suporte robusto a roteamento de mensagens, e ainda por cima, implementa 100% o protocolo AMQP.

Se você já estiver craque nos conceitos de uma EDA e em algum sistema de mensagens, talvez o próximo passo seja começar a usar um framework específico para esse tipo de arquitetura. Recomendo dar uma olhada em três deles:

• NServiceBus - É o mais robusto, maduro e com boa documentação. Tem suporte ao MSMQ e RabbitMQ.
• MassTransit - Funciona bem, porém um pouco experimental e com uma documentação ruim. A curva de aprendizado é um pouco árdua. Também tem suporte ao MSMQ e RabbitMQ.
• EasyNetQ - Só possui suporte ao RabbitMQ, mas é realmente bom e muito (muito mesmo) fácil de aprender. A documentação é boa e eu até contribuí com a tradução dela para português.

Os três frameworks são open source, porém o NServiceBus tem um modelo pago. Por isso, leia bem as questões de licença antes de pensar em implementar um projeto com ele -- você não pode simplesmente sair o usando em produção.

Obrigado por visitar o blog!

download da solução (10 Kb)

MSMQ: Request/Response

Dando continuidade à série de tutoriais básicos sobre como usar o MSMQ com C# para integrar sistemas, neste post vou mostrar o padrão Request/Response ou Request/Reply. É um padrão muito básico e essencial numa arquitetura de integrações usando sistemas de mensagem.

Pré-Requisitos

Assim como no post anterior, você vai precisar ter o MSMQ habilitado e usar o Visual Studio (ou um compilador para .NET).

Prática

Novamente, assim como no tutorial anterior, vamos criar um modelo canônico que será compartilhado pelas aplicações participantes. Crie um projeto do tipo class library com o nome de Eip.Messages em C#. Declare as seguintes interfaces:

public interface IRequestMessage
{
 string Text { get; set; }
}

e

public interface IResponseMessage
{
 string Text { get; set; }
}

Implemente-as como à seguir:

[Serializable]
public class RequestMessageImpl
 : IRequestMessage
{
 public RequestMessageImpl(string text)
 {
  Text = text;
 }
 public string Text { get; set; }
}

e

[Serializable]
public class ResponseMessageImpl
 : IResponseMessage
{
 public ResponseMessageImpl(string text)
 {
  Text = text;
 }
 public string Text { get; set; }
}

Agora, vamos criar o projeto que faz a requisição. Adicione à solução um novo projeto do tipo Console em C# com o nome Eip.Request. Adicione também referências ao projeto Eip.Messages e a System.Messaging. Depois disso, cole o código seguinte dentro do método static void Main(string[] args).

Console.Title = "Request Endpoint";
Console.WindowHeight = 10;
Console.WindowWidth = 50;

// Instancia a fila da requisição (request)
MessageQueue requestQueue = new MessageQueue(@".\private$\request_queue");

// Instancia a fila da resposta (response)
MessageQueue responseQueue = new MessageQueue(@".\private$\response_queue");

// Configura o que será lido das mensagens.
// Para simplificar o exemplo, usamos o SetAll()
MessagePropertyFilter filter = new MessagePropertyFilter();
filter.SetAll();
requestQueue.MessageReadPropertyFilter = filter;
responseQueue.MessageReadPropertyFilter = filter;

// Define qual formatador a fila está usando
responseQueue.Formatter = new BinaryMessageFormatter();

while (true)
{
 Console.WriteLine("Entre com a requisição:");
 string inputRequest = Console.ReadLine();

 if (inputRequest == "sair")
  break;

 // Cria o objeto requisição com a entrada do usuário
 IRequestMessage request = new RequestMessageImpl(inputRequest);

 // Instancia a mensagem que encapsula nossa requisição e define como ela será formatada 
 // para entrar na fila
 Message requestMessage = new Message(request, new BinaryMessageFormatter());

 // Atribui a propriedade de fila de resposta que é onde vamos receber a resposta
 requestMessage.ResponseQueue = responseQueue;
 
 // Envia a requisição
 requestQueue.Send(requestMessage);
 
 // Armazena o Id de correlação
 string correlationId = requestMessage.Id;

 // Bloqueia até receber a mensagem de retorno ou expirar o timeout (5 segundos: new TimeSpan(0, 0, 5))
 // Note que estamos aguardando uma resposta com o Id que relacionamos no passo anteior
 Message responseMessage = responseQueue.ReceiveByCorrelationId(correlationId, new TimeSpan(0, 0, 5));

 // Recupera o objeto resposta
 IResponseMessage responseObject = (IResponseMessage)responseMessage.Body;
 
 Console.WriteLine(responseObject.Text);

 // E a vida segue...
}

Precisamos criar as duas filas que são referenciadas no código acima: .\private$\request_queue e .\private$\response_queue. Crie-as como sugerido no tutorial anterior, caso contrário, a aplicação não vai funcionar.

Inicie a aplicação e entre com uma requisição. Uma exceção será lançada após 5 segundos sem a resposta; vamos consertar isso.

Após as filas terem sido criadas, crie um novo projeto também do tipo Console com o nome Eip.Response para adicioná-lo à solução. Adicione referências ao projeto Eip.Messages e a System.Messaging. Depois disso, cole o código seguinte dentro do método static void Main(string[] args).

Console.Title = "Response Endpoint";
Console.WindowHeight = 10;
Console.WindowWidth = 70;

// Instancia a fila da requisição (request).
// Esta fila deve ser a mesma do lado da requisição -- os dois endpoints (request/response) precisam entrar num acordo aqui
MessageQueue requestQueue = new MessageQueue(@".\private$\request_queue");

// Configura o que será lido das mensagens.
// Para simplificar o exemplo, usamos o SetAll()
MessagePropertyFilter filter = new MessagePropertyFilter();
filter.SetAll();
requestQueue.MessageReadPropertyFilter = filter;

// Temos que trabalhar com algum formato, escolhemos o binário por ser o mais rápido
// É muito comum usar o XmlMessageFormatter por ser legível.
requestQueue.Formatter = new BinaryMessageFormatter();

Console.WriteLine("Aguardando requisições...");

while (true)
{
 // Bloqueia até receber a requisição
 Message requestMessage = requestQueue.Receive();

 // Recupera o objeto requisição
 IRequestMessage requestObject = (IRequestMessage)requestMessage.Body;

 // Processa a requisição e gera um objeto de resposta
 IResponseMessage responseObject = new ResponseMessageImpl(requestObject.Text.ToUpper());

 Console.WriteLine("Requisição recebida: '{0}'", requestObject.Text);

 // Encapsula o objeto resposta dentro de uma mensagem
 Message responseMessage = new Message(responseObject, new BinaryMessageFormatter());
 
 // Passo fundamental! Correlaciona os objetos de resposta e requisição.
 // Sem isso, o outro lado nunca vai receber a resposta
 responseMessage.CorrelationId = requestMessage.Id;

 // Quem define a fila da resposta é quem faz a requisição
 // A responsta poderia ir para qualquer lugar, inclusive um terceito endpoint
 MessageQueue responseQueue = requestMessage.ResponseQueue;

 // Envia a mensagem de resposta (sério?!)
 responseQueue.Send(responseMessage);

 Console.WriteLine("Responsta enviada: '{0}'", responseObject.Text);
}

Inicie as duas aplicações console e veja as requisições sendo respondidas. As respostas nada mais são do que o retorno da string da requisição em maiúsculas.

Pronto! Nossa solução com suporte a requisição e resposta está terminada. Use sua criatividade e faça experimentos com ela.

Balanceamento de Carga Nativo

Uma das vantagens de uma arquitetura baseada em eventos com sistemas de mensagem é a facilidade para o escalonamento. Se você iniciar duas instâncias da aplicação Eip.Receive, vai notar que o consumo das mensagens será dividido por igual em termos de quantidade. Isso é chamado de Round Robin -- cada um consume uma mensagem da fila em modo concorrente e comportado (sem furar a fila).

Conclusão

Note que além do balanceamento de carga nativo, poderíamos serializar as mensagens em algum formato mais facilmente interoperável (XML ou JSON) para que outras aplicações em diferentes plataformas também pudessem participar dessa solução. Reflita também na riqueza de se trocar objetos como mensagens e na possibilidade de definirmos o endereço de resposta dinamicamente. Essas são apenas algumas vantagens desse tipo de arquitetura para integração entre sistemas. É claro que, como qualquer coisa na vida, há desavantagens e vantagens nessa abordagem -- ela não é a solução ideal para todas as integrações.

download da solução (13 Kb)

MSMQ: Tutorial Básico

Conforme dito no post anterior, pretendo iniciar uma pequena série de posts sobre padrões de Arquitetura Orientada a Eventos -- EDA. A ideia desse post é mostrar o básico do básico com foco em como enviar e receber mensagens usando o MSMQ com C# para termos uma base para os padrões.

É importante ressaltar que foi intencional o nível de hard code e falta de abstração. Afinal, não quero nada abstrair aqui; quero mostrar o core da API .NET do MSMQ com C#.

Talvez o primeiro passo antes de discutir algum padrão de integração entre sistemas usando mensagens seja mostrar como se envia e recebe uma mensagem -- as operações mais elementares de um sistema de mensagens.

Pré-Requisitos

Para poder completar os passos seguintes, você precisa ter o MSMQ habilitado no Windows (ele é um recurso nativo do Windows) e o Visual Studio.

Prática

O primeiro passo é criarmos um projeto no Visual Studio do tipo class library em .NET 4 com o nome de Eip.Messages (adotei o prefixo Eip que significa Enterprise Integration Patterns, mas dê o nome que quiser). Esse projeto é como se fosse o nosso modelo canônico. Ou seja, é o modelo que é compartilhado pelas aplicações que participam da integração. É como se fosse um contrato.

Seria viável e mais simples enviar e receber objetos do tipo string ou outros primitivos, mas para uma ideia mais real (e legal) do que podemos fazer com sistemas de mensagem, vamos usar um tipo complexo em forma de interface.

Dentro desse projeto, declare uma interface IMyMessage.

public interface IMyMessage
{
 string Text { get; set; }
}

E a implemente como a seguir:

[Serializable]
public class MyMessageImpl
 : IMyMessage
{
 public MyMessageImpl(string text)
 {
  Text = text;
 }
 public string Text { get; set; }
}

Observação: Em um modelo canônico real, não faz sentido declarar e implementar uma interface no mesmo lugar. Cada aplicação participante poderia implementar da melhor forma ou poderia haver uma implementação padrão separada para possibilitar uma implementação personalizada. Este exemplo está dessa forma por motivos de praticidade e simplicidade.

O segundo passo é adicionarmos um novo projeto do tipo Console chamado Eip.Send à solução e adicionarmos referências a System.Messaging e ao projeto que acabamos de criar. Essa é a aplicação que enviará mensagens -- IMyMessage.

Cole o código à seguir dentro do método static void Main(string[] args).

Console.Title = "Send Endpoint";
Console.WindowHeight = 10;
Console.WindowWidth = 50;

MessageQueue sendQueue = new MessageQueue(@".\private$\my_queue");
   
while (true)
{
 Console.WriteLine("Entre com a mensagem a ser enviada:");
 string inputRequest = Console.ReadLine();

 if (inputRequest == "sair")
  break;

 IMyMessage myEncapsulatedMessage = new MyMessageImpl(inputRequest);

 Message sendMessage = new Message(myEncapsulatedMessage, new BinaryMessageFormatter());

 sendQueue.Send(sendMessage);
 
 Console.WriteLine("mensagem enviada...");
}

Criação de Filas

Antes que nossa aplicação funcione, precisamos criar a fila my_queue. No exemplo acima, estamos fazendo referência a uma fila local (.\) privada (private$) de nome my_queue. É perfeitamente normal criar filas através da aplicação, mas vale criar a fila pelo gerenciador do Windows para colocarmos um pé em questões administrativas da solução.

Para abrir o gerenciado do computador do Windows, execute compmgmt.msc e Uma janela como a seguir deve abrir.

Dê um clique com o botão direito do mouse sobre Private Queues abaixo de Message Queuing que fica dentro de Services and Applications. Escolha New e depois Private Queue. Entre com o nome my_queue mantendo o checkbox Transactional desabilitado e clique em Ok logo após.

Pronto, você acabou de criar uma fila privada não transacional. Se você achou um saco fazer isso, dê uma olhada na classe System.Messaging.MessageQueue e em seus métodos estáticos MessageQueue.Exists e MessageQueue.Create.

A aplicação que envia mensagens já está pronta para ser usada. Inicie-a para dar uma olhada. Ela pede que você entre com uma mensagem e a envia para a fila. Digite qualquer coisa (menos "sair") e dê um Enter.

Meio chato até agora, não? Vamos criar o terceiro projeto que também é do tipo Console com o nome de Eip.Receive e adicionar as mesmas referências do projeto anterior. Essa aplicação vai consumir as mensagens da fila my_queue. Assim como na aplicação anterior, cole o código à seguir dentro do método static void Main(string[] args).

Console.Title = "Receive Endpoint";
Console.WindowHeight = 10;
Console.WindowWidth = 70;

MessageQueue myQueue = new MessageQueue(@".\private$\my_queue");
myQueue.Formatter = new BinaryMessageFormatter();

Console.WriteLine("Aguardando mensagens...");

while (true)
{
 Message receivedMessage = myQueue.Receive();
 IMyMessage myEncapsulatedMessage = (IMyMessage)receivedMessage.Body;
 Console.WriteLine("Mensagem recebida: '{0}'", myEncapsulatedMessage.Text);
}

Uma forma que facilita iniciar as aplicações de forma independente é ir nas propriedades da solução (não projeto) e na opção Startup Project, selecionar Current Selection.

Dessa forma, basta selecionar o projeto e dar um CTRL+F5 ou F5 se quiser debugar. Experimente e faço-o sobre Eip.Receive. Você deve notar que a mensagem que digitamos em alguns passos anteriores vai aparecer na tela.

Conclusão

A parte mais interessante dessa solução é o desacoplamento, principalmente temporal. Quem envia a mensagem não sabe quem recebe, se recebe e quando recebe. E quem recebe a mensagem não sabe quem a enviou e quando a enviou. Faça alguns testes iniciando as aplicações ao mesmo tempo, uma antes da outra, apenas uma de cada vez, etc.

download da solução (11 Kb)

MSMQ Multicast - Múltiplas Interfaces de Rede

O MSMQ tem um problema quando tentamos receber mensagens enviadas como multicast em uma máquina com mais de uma placa de rede -- tanto a máquina receptora quanto a que envia as mensagens. O MSMQ escolhe aleatoriamente uma placa que pode não ser a placa correta para receber as mensagens multicast. Contudo, há uma maneira de contornar isso.

Insira uma chave REG_SZ em HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\MSMQ\Parameters\MulticastBindIP com o IP da sua placa de rede que se conecta com o serviço de enfileiramento de outras máquinas. A chave HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\MSMQ\Parameters\BindInterfaceIP é para ambientes clusterizados.

Após ter adicionado a chave ou chaves, reinicie os serviços de enfileiramento para os novos parâmetros fazerem efeito:

C:\> net stop msmq
C:\> net start msmq

O serviço MSMQTriggers precisará ser reiniciado (automaticamente) também pois depende do MSMQ.


Referência:

• http://technet.microsoft.com/en-us/library/cc756156(WS.10).aspx