• Página web del proyecto
• Guía del framework

Implementar una capa de mensajes con ZeroMQ:

1. Escoger tipo de transporte:
   • TCP, basado en red: tcp://hostname:port
   • INPROC, intraproceso (inter-thread/greenlet): inproc://name
   • IPC, interproceso: ipc:///tmp/filename
   • MULTICAST, multicast via PGM (con OpenPGM y sólo para PUB-SUB): pgm://interface:address:port y epgm://interface:address:port
2. Indicar la infraestructura:
   • Queue, para el request/response messaging pattern
   • Forwarder, para el publish/subscribe messaging pattern.
   • Streamer, para un pipelined paralellized messaging pattern.
3. Escoger el ptarón de mensaje:
   • REQUEST/REPLY, bidireccional, basado en estados y con balanceador de carga.
   • PUBLISH/SUBSCRIBE, publica a varios receptores de golpe.
   • UPSTREAM/DOWNSTREAM, distribución de los datos en una pipeline.
   • PAIR, comunicación exclusiva entre pares.

Pasos para desarrollar:

1. Obtener el contexto de ZeroMQ.
2. A partir del contexto definir los sockets necesarios a partir de la arquitectura que usarán.
3. Si realizan la acción de escucha (elemento estable) llamaremos al método bind del socket. Si empezarán enviando (elemento dinámico) al connect.

El contexto es el contenedor de los sockets para un proceso. Para inicializar el contexto se usa zmq_ctx_new() y para eliminarlo zmq_ctx_destroy().

Propiedades de la librería:

• Podemos conectar un gran número de clientes a un servidor.
• Los clientes guardarán los paquetes hasta que el servidor se inicie.
• Podemos conectar un cliente a tantos servidores como queramos, los mensajes se repartirán entre ellos.

Notas:

• Cuando indicamos * como dirección nos estamos refiriendo a que tome todas las interfaces. Por ejemplo: http://*:5555
• Si un servidor deja de funcionar y el cliente está aún conectado, el funcionamiento de este último no será estable.
• ZeroMQ no se encarga de la codificación de mensajes. Podremos enviarlo en el formato que queramos, por ejemplo simple JSON o de forma binaria con BSON, Protocol Buffers o Thrift.

Esta arquitectura consiste en que el cliente envía un mensaje con zmq_send() y luego recibe con zmq_recv(), el servidor lo hace en el orden contrario. Estas acciones se lleban a cabo en loop y sin poder cambiarse de orden.

Servidor:

import zmq
import time
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")
while True:
    message = socket.recv()
    # Do some 'work'
    socket.send("World")

Cliente:

import zmq
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect ("tcp://localhost:5555")
for request in range (10):
    socket.send ("Hello")
    #  Get the reply.
    message = socket.recv()
    print "Received reply ", request, "[", message, "]"

Es unidireccional. En este patrón un servidor publica sin importar quien recibe. Enviará a todos los clientes y serán estos quienes filtren.
Por ejemplo el siguiente código envía eventos de los distintos participantes…

import zmq
from random import choice
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:5000")
countries = ['netherlands','brazil','germany','portugal']
events = ['yellow card', 'red card', 'goal', 'corner', 'foul']
while True:
    msg = choice( countries ) +" "+ choice( events )
    print "->",msg
    socket.send( msg )

… Y el siguiente recibe de Netherlands y Germany.

import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:5000")
socket.setsockopt(zmq.SUBSCRIBE, "netherlands")
socket.setsockopt(zmq.SUBSCRIBE, "germany")
while True:
    print  socket.recv()

Notas:

• Un subscriber puede conectarse a tantos publishers como sea necesario.
• Si un publisher no tiene subscribers los mensajes quedarán desechados.
• TCP con esta estructura es lento.

Funciona como el Rep-Req sólo que en este no hay una respuesta desde el otro nodo. Es perfecto para los mensajes en paralelo. Imaginemos la siguiente estructura:

En esta…

• Un proceso ventilador (productor) produce tareas que pueden ser realizadas en paralelo.
• Las tareas la reciben los workers.
• Existe un sink que recoge los resultados de los workers.

Ventilador:

import zmq
import random
import time
context = zmq.Context()
 
sender = context.socket(zmq.PUSH)
sender.bind("tcp://*:5557")
 
sink = context.socket(zmq.PUSH)
sink.connect("tcp://localhost:5558")
 
# The first message is "0" and signals start of batch
sink.send('0')
 
# Send 100 tasks
total_msec = 0
for task_nbr in range(100):
    workload = random.randint(1, 100)
    total_msec += workload
    sender.send(str(workload))

Worker:

import sys
import time
import zmq
 
context = zmq.Context()
 
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")
 
sender = context.socket(zmq.PUSH)
sender.connect("tcp://localhost:5558")
 
while True:
    s = receiver.recv()
    sys.stdout.write('.')
    time.sleep(int(s)*0.001)
    sender.send('')

Sink:

import sys
import time
import zmq
 
context = zmq.Context()
 
receiver = context.socket(zmq.PULL)
receiver.bind("tcp://*:5558")
 
# Wait for start of batch
s = receiver.recv()
 
tstart = time.time()
# Process 100 confirmations
total_msec = 0
for task_nbr in range(100):
    s = receiver.recv()
    if task_nbr % 10 == 0:
        sys.stdout.write(':')
    else:
        sys.stdout.write('.')
tend = time.time()
print "Total elapsed time: %d msec" % ((tend-tstart)*1000)

Estos son una comunicación uno a uno:

import zmq
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.bind("tcp://127.0.0.1:5555")

import zmq
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.connect("tcp://127.0.0.1:5555")

Podemos configurar un nodo como device. Esto permitirá mejorar el rendimiento del sistema. Por ejemplo una queue para un Request-Reply:

Cuando enviamos un mensaje y no hay receptor, lo que hace ZeroMQ es guardarlo en memoria hasta que el receptor esté activo.
Para evitar problemas de memoria ZeroMQ provee de la opción de socket high water mark (zmq.HWM). Si

sock = context.socket(zmq.PUSH)
sock.setsockopt(zmq.HWM, 1000)
sock.setsockopt(zmq.SWAP, 200*2**10)
sock.connect(sys.argv[1])
while True:
    sock.send(sys.argv[1] + ':' + time.ctime())

• Saber la versión con zmq.zmq_version().

sock.send(part1, zmq.SNDMORE)
sock.send(part2, zmq.SNDMORE)
sock.send(part3, zmq.SNDMORE)
sock.send(final)

more = True
parts = []
while more:
    parts.append(sock.recv())
    more = sock.getsockopt(zmq.RCVMORE)

Productor:

import zmq
import random
import time
 
context = zmq.Context()
 
sender = context.socket(zmq.PUSH)
sender.bind("tcp://*:5557")
sender.bind("tcp://*:5558")
 
print "Sending tasks to workers"
random.seed()
total_msec = 0
for task_nbr in range(100):
    workload = random.randint(1, 100)
    total_msec += workload
    sender.send(str(workload))
 
print "Total expected cost: %s msec" % total_msec
time.sleep(1)

Consumidor:

import sys
import time
import zmq
 
context = zmq.Context()
 
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")
 
while True:
    s = receiver.recv()
    sys.stdout.write('.'+s)
    sys.stdout.flush()
    # Do the work
    time.sleep(int(s)*0.001)