Tabla de Contenidos

ZeroMQ

ZeroMQ

Uso de la librería

Implementar una capa de mensajes con ZeroMQ:

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
Indicar la infraestructura:
- Queue, para el request/response messaging pattern
- Forwarder, para el publish/subscribe messaging pattern.
- Streamer, para un pipelined paralellized messaging pattern.
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:

Obtener el contexto de ZeroMQ.
A partir del contexto definir los sockets necesarios a partir de la arquitectura que usarán.
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.

Conceptos, patrones y arquitecturas

Request-Reply

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, "]"

Publish-Subscribers (Pub\Sub)

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.
Para realizar una subscripción a todos los mensajes haremos:

socket.setsockopt(zmq.SUBSCRIBE, "")

Pipeline

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)

Paired sockets

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")

Devices

Podemos configurar un nodo como device. Esto permitirá mejorar el rendimiento del sistema.

Queue

Se usa para mediar entre un REQ/REP:

Para poder llevarlo a cabo se utilizan dos nuevos tipos de sockets zmq.ROUTER y zmq.DEALER en el nodo queue. El socket ROUTER recibe un mensaje de una conexión REQ y añade un ID al inicio de dicho paquete, luego es enviado por el DEALER. El DEALER escoge una conexión REP y lo envía, cuando este es procesado es devuelto otra al ROUTER y enviado al socket REQ que inició la conexión. Todo esto es posible (y rompe el patrón REQ\REP) mediante el ID que el ROUTER inserta.
REQ:

import sys
import zmq
context = zmq.Context()
for x in xrange(10):
    sock = context.socket(zmq.REQ)
    sock.connect(sys.argv[1])
    print 'REQ is', x,
    sock.send(str(x))
    print 'REP is', sock.recv()

REP:

import sys
import zmq
context = zmq.Context()
while True:
    sock = context.socket(zmq.REP)
    sock.connect(sys.argv[1])
    x = sock.recv()
    print 'REQ is', x,
    reply = 'x-%s' % x
    sock.send(reply)
    print 'REP is', reply

QUEUE:

import sys
import zmq
context = zmq.Context()
s1 = context.socket(zmq.ROUTER)
s2 = context.socket(zmq.DEALER)
s1.bind(sys.argv[1])
s2.bind(sys.argv[2])
zmq.device(zmq.QUEUE, s1, s2)

Forwarding

Media entre un grupo de nodos push\sub. Como este conjunto no se bloquea no es necesario usar un tipo de socket especial:

import sys
import zmq
context = zmq.Context()
s1 = context.socket(zmq.SUB)
s2 = context.socket(zmq.PUB)
s1.bind(sys.argv[1])
s2.bind(sys.argv[2])
s1.setsockopt(zmq.SUBSCRIBE, '')
zmq.device(zmq.FORWARDER, s1, s2)

Ahora podemos conectarle tantos publishers y subscribers como queramos.

Streaming

En este caso es útil para los nodos PUSH\PULL.

context = zmq.Context()
s1 = context.socket(zmq.PULL)
s2 = context.socket(zmq.PUSH)
s1.bind("tcp://*:5550")
s2.bind("tcp://*:5551")
zmq.device(zmq.STREAMER, s1, s2)

import zmq
import random
import time
context = zmq.Context()
 
sender = context.socket(zmq.PUSH)
sender.connect("tcp://localhost:5550")
 
total_msec = 0
for task_nbr in range(100):
    workload = random.randint(1, 100)
    sender.send(str(workload))

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

Crear tu propio Streamer & Forwarder

Como los dos son unidireccionales, lo único que tendremos que hacer es copiar los datos de un socket al otro:

import sys
import zmq
 
def zmq_streaming_device(a, b):
    while True:
        msg = a.recv()
        more = a.getsockopt(zmq.RCVMORE)
        if more:
            b.send(msg, zmq.SNDMORE)
        else:
            b.send(msg)
        sys.stdout.write('.')
        sys.stdout.flush()
 
context = zmq.Context()
s1 = context.socket(zmq.PULL)
s2 = context.socket(zmq.PUSH)
s1.bind("tcp://*:5550")
s2.bind("tcp://*:5551")
zmq_streaming_device(s1, s2)

Patrones Request/Reply

Patrones Publish/Subscribe

Avanzado

Control de flujo

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 la añadimos a setsockopt podremos indicar cuantos mensajes quedarán guardados en memoria.
Con la opción zmq.SWAP podemos indicar un tamaño de disco donde guardar más mensajes una vez el HWM sea superado.

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())

Notas

Saber la versión con zmq.zmq_version().
Para realizar conexiones no bloqueantes los usaremos con greenlets.
Para instalar la librería en el sistema: apt-get install libzmq-dev.

Instalar en Ubuntu

Instalación de la librería del sistema:

$ sudo apt-get install python-dev libzmq-dev

Instalación en Python:

$ sudo pip install pyzmq

Recuerda que para utilizar pip has de tener instalado el paquete python-pip

Instalar en Windows

Necesitarás agregar el paquete VS2010 C++ redistributable.

Como...

Enviar\recibir varios paquetes

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)

Sistema Productor\Consumidor

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)

Enviar no bloqueante

Pero cuidado, el objeto sender no se elminará hasta que los datos hayan sido enviados.

sender.send(jdata, zmq.NOBLOCK)

Programming

Herramientas de usuario

Herramientas del sitio