ZeroMQ: Request/Reply com Balanceamento de Carga

Se você já entendeu o básico para fazer um request/reply com dois componentes; está na hora de elaborar um pouco mais o negócio. Vamos criar um request/reply com um balanceador no meio. Na verdade, nós vamos trabalhar imaginando escalar nossa solução horizontalmente com mais de um serviço que responda as requisições, afinal é moda e essa moda é boa ;)

O exemplo a seguir, é baseado nesse trecho do guia do ZeroMQ: http://zguide.zeromq.org/page:all#Shared-Queue-DEALER-and-ROUTER-sockets (figura 16, a segunda desse tópico).

Código

Se você seguiu todo os passos dos posts anteriores ou já conhece como configurar seu projeto com ZeroMQ, ótimo. Se você não sabe do que estou falando, procure pelos posts sobre ZeroMQ que ensinam como incluir o ZeroMQ em seu projeto. Se você esta sem paciência, use o comando seguinte do Package Manager Console: PM> Install-Package clrzmq -pre

Vamos lá, crie 3 projetos do tipo console, .NET 4.5, C#.

O primeiro projeto é o do serviço que realiza a resposta, ou reply, ou servidor. Nós vamos chamá-lo de Worker, porque é o componente que faria o processamento de fato; o trabalho pesado. Eu dei o nome dele de zeromq.lab.worker. Os nomes das variáveis, mensagens, etc. estão em inglês nesse exemplo porque era uma POC antiga e às vezes cismo de aderir ao inglês -- mantive para não ter que traduzir tudo.

Direto ao ponto: use o trecho a seguir para seu Worker:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using ZeroMQ;

namespace zeromq.lab.worker
{
 class Worker
 {
  static void Main(string[] args)
  {
   Console.Title = "Worker";
   Console.WindowHeight = 10;
   Console.WindowWidth = 70;

   using (var context = ZmqContext.Create())
   using (var replier_sckt = context.CreateSocket(SocketType.REP))
   {
    replier_sckt.Connect("tcp://127.0.0.1:5001");
    Message("connected...");

    while (true)
    {
     ZmqMessage request = replier_sckt.ReceiveMessage();
     string request_payload = Encoding.Unicode.GetString(request[0]);

     Message("request received: {0}", request_payload);

     string reply_payload = request_payload.ToUpper().Replace("-", string.Empty);

     ZmqMessage resposta = new ZmqMessage(new List<byte[]> 
     {
      Encoding.Unicode.GetBytes(reply_payload)
     });

     replier_sckt.SendMessage(resposta);
     Message("reply sent: {0}", reply_payload);
    }
   }
  }

  static void Message(string message, params object[] arg)
  {
   Console.WriteLine(
    string.Format("{0} - {1}",
     DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff"),
     message), arg);
  }
 }
}

O segundo projeto é o projeto do cliente. Num cenário real de uma aplicação web, é este componente que estaríamos elaborando. Mas como um console é mais rápido para se fazer, vamos usá-lo. Novamente, use o trecho a seguir para seu cliente.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using ZeroMQ;

namespace zeromq.lab.client
{
 class Client
 {
  static void Main(string[] args)
  {
   Console.Title = "Client";
   Console.WindowHeight = 10;
   Console.WindowWidth = 50;

   using (ZmqContext ctx = ZmqContext.Create())
   using (ZmqSocket request_sckt = ctx.CreateSocket(SocketType.REQ))
   {
    request_sckt.Connect("tcp://127.0.0.1:5000");

    while (true)
    {

     string request_id = Guid.NewGuid().ToString();

     ZmqMessage request = new ZmqMessage(new List<byte[]> 
     {
      Encoding.Unicode.GetBytes(request_id)
     });


     request_sckt.SendMessage(request);
     Message("request sent");

     ZmqMessage reply = request_sckt.ReceiveMessage();
     Message("reply received: {0}", Encoding.Unicode.GetString(reply[0]));

     Thread.Sleep(new Random().Next(5000));
    }

   }

  }

  static void Message(string message, params object[] arg)
  {
   Console.WriteLine(
    string.Format("{0} - {1}",
     DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff"),
     message), arg);
  }
 }
}

O último projeto é o nosso broker, ou balanceador de carga, mais especificamente. Este componente é responsável por repassar as requisições aos serviços de forma uniforme e circular. Essa estratégia de distribuição se chama round robin. Use o trecho a seguir de código para completar nosso exemplo.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using ZeroMQ;

namespace zeromq.lab.broker
{
 class Broker
 {
  static void Main(string[] args)
  {
   Console.Title = "THE BROKER";
   Console.WindowHeight = 10;
   Console.WindowWidth = 50;

   using (ZmqContext context = ZmqContext.Create())
   using (ZmqSocket
    frontend = context.CreateSocket(SocketType.ROUTER),
    backend = context.CreateSocket(SocketType.DEALER))
   {
    frontend.Bind("tcp://*:5000");
    backend.Bind("tcp://*:5001");

    Message("ports binded");

    frontend.ReceiveReady += (sender, e) => 
    {
     ZmqMessage request = e.Socket.ReceiveMessage();
     backend.SendMessage(request);
    };

    backend.ReceiveReady += (sender, e) =>
    {
     ZmqMessage reply = e.Socket.ReceiveMessage();
     frontend.SendMessage(reply);
    };

    Poller poller = new Poller(new List<ZmqSocket>
    { 
     frontend, 
     backend
    });

    while (true)
    {
     poller.Poll();
     Message("polled");
    }
   } 
  }

  static void Message(string message, params object[] arg)
  {
   Console.WriteLine(
    string.Format("{0} - {1}", 
     DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff"), 
     message), arg);
  }
 }
}

Agora, inicie mais de uma instância do projeto do Worker (para poder ver a distribuição de carga entre eles). Inicie o broker e o cliente em seguida.

Pronto! Espero que tenha gostado desse exemplo.