Objetivo

Nesta parte veremos como realizar integração através de MSMQ. MSMQ é uma tecnologia baseada em filas de propriedade da Microsoft (concorrente direta do Websphere MQ, anteriormente conhecido como MQSeries, da IBM).

Esse método de integração é popularmente conhecido como “mensageria”. É muito comum no mercado financeiro. Não estaremos abordando aqui todos os possíveis mecanismos de integração, apenas o que está sendo explorado nesta série que é a integração de grande volume de dados entre duas aplicações.

Este método também tem algumas características interessantes:

• Ele pode ser transacional, ou seja, existe uma garantia de entrega e processamento das mensagens
• Devido a esta característica transacional, a mensagem só pode ser entregue de forma atômica (inteira, nunca em pedaços)
• Nos sistemas de mensageria, o sistema origem sempre entrega a mensagem para a fila de mensagens e o sistema destino pega a informação desta fila, o que faz com que exista um menor acoplamento entre eles (óbvio que eles precisam conhecer a mesma fila). O sistema destino também não precisa estar disponível para que a mensagem seja entregue (isso não quer dizer que ele não precise estar disponível para que a mensagem seja processada)
• O sistema de filas garante a entrega da mensagem, porém, não faz nenhum tratamento em relação ao formato da mensagem. Desta forma, precisa existir um formato acordado entre os sistemas. Por essa razão, sistemas de mensageria são muito interessantes se usados em conjunto com barramentos de serviço (message brokers), pois esse trabalho de transformação tanto técnica quanto no conteúdo da mensagem (enriquecimento, por exemplo) pode ser realizado por uma ferramenta especialista nisso, tirando código de integração de dentro dos sistemas

Todas as características acima aplicam-se tanto ao MSMQ quanto ao Websphere MQ. No próximo post, falarei sobre Websphere MQ.

Setup

No nosso exemplo, precisaremos apenas criar as filas e configurar o MSMQ na máquina servidora.

O MSMQ é um componente do Windows. Ele não vem instalado por Default. No Windows 7, para instalá-lo basta seguir os passos:

• Painel de Controle | Programas e Recursos | Ativar ou desativar recursos do Windows
• Habilitar todos os itens abaixo de Servidor do MSMQ (Microsoft Message Queue)

Após clicar em Ok, o MSMQ será instalado.

Após a instalação, será necessário configurar as filas. Para isso, no Windows Explorer, clique com o botão direito no computador em seguida em “Gerenciar”. Em serviços e aplicativos aparecerá “Enfileiramento de mensagens”.

Em “filas privativas”, clique com o botão direito e nova. Em seguida criaremos duas filas (transacionais): integrationtests.in, integrationtests.out.

Servidor

O servidor é bastante simples, segue o código:

        public override bool  Execute()
        {
            Log.LogMessage("Starting MSMQ Server");

            outputQueue = new MessageQueue(OutputQueueName);

            inputQueue = new MessageQueue(InputQueueName);
            inputQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(queue_ReceiveCompleted);
            inputQueue.BeginReceive();

            while (true)
                Thread.Sleep(250);                        
        }

        private void queue_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e)
        {
            
            Log.LogMessage("Message received " + messageCount.ToString());
            messageCount++;

            Message resp = new Message();
            resp.BodyStream = new MemoryStream();
            
            util.ProcessClientBigRequest(ConnString, e.Message.BodyStream, resp.BodyStream, false, null);
            resp.BodyStream.Seek(0, SeekOrigin.Begin);
            
            outputQueue.Send(resp, MessageQueueTransactionType.Single);

            inputQueue.BeginReceive();
        }

A idéia aqui é ouvir o evento “ReceiveCompleted” da fila, já disparar um BeginReceive. Isso significa que a cada mensagem que pingar na fila, o método queue_ReceiveCompleted é chamado.

Sempre que chega uma nova mensagem, o servidor processa o request, e já armazena a resposta na própria stream da mensagem de resposta, passando a mesma como parâmetro no método ProcessClientBigRequest.

A mensagem com a resposta é colocada na fila de saída.

Cliente

O cliente segue a mesma filosofia do servidor:

        public override bool Execute()
        {
            Stopwatch watch = new Stopwatch();
            watch.Start();

            Log.LogMessage("Starting MSMQ transfer with " + TotalBatches.ToString() + " batchs with " + BatchSize.ToString() + " items each");            

            inputQueue = new MessageQueue(InputQueueName);
            inputQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(inQueue_ReceiveCompleted);
            inputQueue.BeginReceive();

            MessageQueue queue = new MessageQueue(OutputQueueName);            
            int count = 1;            
            for (int i = 0; i < TotalBatches; i++)
            {
                Message msg = new Message();
                msg.BodyStream = new MemoryStream();
                
                util.GenerateBigRequest(msg.BodyStream, false, count, count + (BatchSize - 1));
                msg.BodyStream.Seek(0, SeekOrigin.Begin);
                queue.Send(msg, MessageQueueTransactionType.Single);                

                count += BatchSize;
                messageCount++;
            }

            while (!finished)
                Thread.Sleep(250);

            watch.Stop();
            Log.LogMessage("Total processing time: " + watch.Elapsed.TotalSeconds.ToString("0.00") + " seconds");

            return true;
        }

        private void inQueue_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e)
        {
            inputQueue.BeginReceive();  

            Log.LogMessage("Received message. Waiting for " + messageCount.ToString() + " messages");

            messageCount--;
            util.ImportarStream(ConnString, e.Message.BodyStream);            

            if (messageCount <= 0)
                finished = true;            
        }

O cliente também inicializa sua fila de entrada e começa a esperar as mensagens de resposta. Em seguida, gera os lotes de requisições. Para cada mensagem recebida, o contador messageCount é decrementado, para saber a hora de parar de receber mensagens e calcular os tempos.

Código fonte

O código fonte está disponível no git hub: https://github.com/ericlemes/IntegrationTests.

Conclusão

Surpreendentemente este método foi um dos mais performáticos sem o uso de filas transacionais, sempre seguindo essa idéia de quebrar em pequenos lotes. Se cada requisição gerar uma mensagem, ele também fica bastante prejudicado (como nos web services).

Usando filas transacionais, a performance caiu bastante. Essa implementação ainda precisa ser revista. Acredito que dê para melhorar significativamente a performance dela (ver post: Integrações entre sistemas – Parte 12 – MSMQ, revisitado).

O principal ganho aqui é que os processadores das duas máquinas estão trabalhando simultaneamente, devido à natureza assíncrona da comunicação. No momento que o cliente está gerando sua segunda requisição, já está processando a primeira resposta, ou seja, os intervalos em que existe espera pela rede, as respostas estão sendo processadas, tanto no cliente quanto no servidor.