ZeroMQ (también escrito ØMQ o ZMQ) es una librería de mensajería asíncrona que se integra directamente en tus aplicaciones. A diferencia de RabbitMQ o Kafka, no hay un servidor intermedio: los procesos se conectan entre sí mediante sockets de ZeroMQ, que se encargan de reconectar, encolar y repartir los mensajes. En este tutorial instalarás ZeroMQ y su binding de Python, pyzmq, en Ubuntu 24.04 y programarás los tres patrones básicos: petición-respuesta, publicación-suscripción y reparto de trabajo con push-pull.

Requisitos previos

Para seguir esta guía necesitas:

  • Un servidor o equipo con Ubuntu 24.04 LTS, por ejemplo un VPS de CubePath.
  • Un usuario no root con privilegios sudo.
  • Conocimientos básicos de Python 3.
  • Poder abrir dos o tres sesiones de terminal a la vez (varias sesiones SSH o tmux).

Conceptos básicos de ZeroMQ

Antes de programar, conviene tener claras cuatro ideas:

  • Contexto: el objeto zmq.Context gestiona los hilos de E/S internos. Se crea uno por proceso.
  • Sockets tipados: cada socket tiene un tipo (REQ, REP, PUB, SUB, PUSH, PULL...) que determina con qué otros tipos puede hablar y cómo reparte los mensajes.
  • bind y connect: el extremo estable (el servidor) hace bind en una dirección y los extremos variables hacen connect. El orden de arranque no importa: si el cliente arranca antes, ZeroMQ reintenta la conexión solo.
  • Mensajes, no bytes: ZeroMQ entrega mensajes completos, nunca fragmentos, así que no hace falta delimitar los datos.

Los transportes más usados son tcp:// (entre máquinas), ipc:// (entre procesos de la misma máquina, mediante un socket Unix) e inproc:// (entre hilos del mismo proceso).

Estos son los patrones que verás en la guía:

PatrónSocketsUso típico
Petición-respuestaREQ con REPLlamadas tipo RPC, un cliente espera la respuesta
Publicación-suscripciónPUB con SUBDifundir eventos a muchos receptores, filtrando por prefijo
PipelinePUSH con PULLRepartir tareas entre trabajadores y recoger resultados

Paso 1: Instalar ZeroMQ y pyzmq

Ubuntu 24.04 incluye la librería libzmq 4.3.5 y el binding de Python en sus repositorios. Instala el paquete de Python, que arrastra la librería como dependencia:

sudo apt update
sudo apt install python3-zmq

Comprueba las versiones de la librería y del binding:

python3 -c "import zmq; print('libzmq', zmq.zmq_version(), '| pyzmq', zmq.pyzmq_version())"

libzmq 4.3.5 | pyzmq 24.0.1

Crea un directorio para los ejemplos:

mkdir -p ~/zmq-ejemplos
cd ~/zmq-ejemplos

Paso 2: Patrón petición-respuesta (REQ/REP)

En este patrón, el cliente envía una petición y espera la respuesta antes de enviar la siguiente. El servidor recibe, procesa y responde. Crea el servidor:

nano ~/zmq-ejemplos/servidor_rep.py

import zmq

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://127.0.0.1:5555")
print("Servidor escuchando en el puerto 5555")

while True:
    peticion = socket.recv_json()
    resultado = peticion["a"] + peticion["b"]
    print(f"Recibido {peticion}, respondo {resultado}")
    socket.send_json({"resultado": resultado})

Crea el cliente. Usa un tiempo de espera para no quedarse bloqueado para siempre si el servidor no responde:

nano ~/zmq-ejemplos/cliente_req.py

import zmq

context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.setsockopt(zmq.RCVTIMEO, 3000)  # 3 segundos
socket.setsockopt(zmq.LINGER, 0)
socket.connect("tcp://127.0.0.1:5555")

for i in range(3):
    socket.send_json({"a": i, "b": 10})
    try:
        respuesta = socket.recv_json()
        print(f"{i} + 10 = {respuesta['resultado']}")
    except zmq.Again:
        print("Sin respuesta del servidor")
        break

socket.close()
context.term()

Arranca el servidor en una terminal:

python3 ~/zmq-ejemplos/servidor_rep.py

Y el cliente en otra:

python3 ~/zmq-ejemplos/cliente_req.py

0 + 10 = 10
1 + 10 = 11
2 + 10 = 12

El servidor muestra cada petición recibida. Si ejecutas el cliente varias veces, o varios clientes a la vez, el socket REP los atiende por turnos. Detén el servidor con Ctrl+C.

Paso 3: Patrón publicación-suscripción (PUB/SUB)

Un publicador envía mensajes sin saber quién los recibe, y cada suscriptor indica qué prefijos le interesan. El filtrado se hace por el inicio del mensaje, así que es habitual empezar cada mensaje con un tema. Crea el publicador:

nano ~/zmq-ejemplos/publicador.py

import random
import time

import zmq

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:5556")

ciudades = ["madrid", "barcelona", "sevilla"]

while True:
    ciudad = random.choice(ciudades)
    temperatura = round(random.uniform(10, 35), 1)
    socket.send_string(f"clima.{ciudad} {temperatura}")
    time.sleep(0.5)

Crea un suscriptor que solo reciba los mensajes de Madrid:

nano ~/zmq-ejemplos/suscriptor.py

import zmq

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:5556")
socket.subscribe("clima.madrid")

for _ in range(5):
    tema, valor = socket.recv_string().split()
    print(f"{tema}: {valor} °C")

Arranca el publicador en una terminal y el suscriptor en otra:

python3 ~/zmq-ejemplos/publicador.py

python3 ~/zmq-ejemplos/suscriptor.py

clima.madrid: 27.4 °C
clima.madrid: 15.9 °C
clima.madrid: 31.2 °C
clima.madrid: 22.8 °C
clima.madrid: 12.6 °C

Solo llegan los mensajes de Madrid: el resto se filtran. Para recibir todos los temas, usa socket.subscribe("").

Ten en cuenta dos comportamientos de este patrón:

  • Sin historial: un suscriptor solo recibe lo que se publica después de conectarse. Los mensajes anteriores se pierden, y los primeros milisegundos tras conectar también (el llamado slow joiner).
  • Sin control de entrega: si un suscriptor es lento y se llena su cola (1.000 mensajes por defecto, ajustable con zmq.SNDHWM), el publicador descarta mensajes para ese suscriptor en lugar de bloquearse.

Si necesitas que no se pierda ningún mensaje, usa un broker con persistencia como RabbitMQ o Kafka. Detén el publicador con Ctrl+C.

Paso 4: Patrón pipeline (PUSH/PULL)

Este patrón reparte tareas entre varios trabajadores en rueda (round robin) y recoge los resultados en un colector. Tiene tres piezas: un distribuidor que envía tareas, los trabajadores que las procesan y un colector que recibe los resultados.

Crea el trabajador. Recibe tareas por el puerto 5557 y envía los resultados al colector por el 5558:

nano ~/zmq-ejemplos/trabajador.py

import os
import time

import zmq

context = zmq.Context()

tareas = context.socket(zmq.PULL)
tareas.connect("tcp://127.0.0.1:5557")

resultados = context.socket(zmq.PUSH)
resultados.connect("tcp://127.0.0.1:5558")

pid = os.getpid()
while True:
    duracion_ms = tareas.recv_json()["ms"]
    time.sleep(duracion_ms / 1000)
    resultados.send_json({"pid": pid, "ms": duracion_ms})

Crea el colector, que espera tantos resultados como tareas haya y cuenta cuántas ha hecho cada trabajador:

nano ~/zmq-ejemplos/colector.py

import collections
import time

import zmq

TOTAL = 20

context = zmq.Context()
socket = context.socket(zmq.PULL)
socket.bind("tcp://127.0.0.1:5558")

inicio = time.monotonic()
por_trabajador = collections.Counter()
for _ in range(TOTAL):
    resultado = socket.recv_json()
    por_trabajador[resultado["pid"]] += 1

print(f"{TOTAL} tareas en {time.monotonic() - inicio:.2f} s")
for pid, cuenta in por_trabajador.items():
    print(f"  trabajador {pid}: {cuenta} tareas")

Crea el distribuidor. Espera a que pulses Intro para dar tiempo a que los trabajadores se conecten; si no, el primero en conectarse se llevaría todas las tareas:

nano ~/zmq-ejemplos/distribuidor.py

import zmq

TOTAL = 20

context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.bind("tcp://127.0.0.1:5557")

input("Pulsa Intro cuando los trabajadores estén listos... ")
for _ in range(TOTAL):
    socket.send_json({"ms": 500})
print(f"{TOTAL} tareas enviadas")

Arranca el colector en una terminal:

python3 ~/zmq-ejemplos/colector.py

En una segunda terminal, lanza cuatro trabajadores en segundo plano:

for i in 1 2 3 4; do python3 ~/zmq-ejemplos/trabajador.py & done

En esa misma terminal, arranca el distribuidor y pulsa Intro:

python3 ~/zmq-ejemplos/distribuidor.py

El colector muestra el resultado:

20 tareas en 2.51 s
  trabajador 51233: 5 tareas
  trabajador 51234: 5 tareas
  trabajador 51235: 5 tareas
  trabajador 51236: 5 tareas

Veinte tareas de medio segundo han tardado unos 2,5 segundos en lugar de 10, porque se han repartido a partes iguales entre los cuatro trabajadores. Detén los trabajadores:

kill %1 %2 %3 %4

Paso 5: Comunicar varias máquinas

Los ejemplos usan 127.0.0.1 para no exponer nada. Para conectar procesos en servidores distintos, el lado que hace bind debe escuchar en la IP privada, por ejemplo socket.bind("tcp://10.0.0.11:5555"), y los clientes conectarse a esa dirección.

ZeroMQ no cifra ni autentica por defecto. Abre el puerto solo a las máquinas que lo necesitan:

sudo ufw allow from 10.0.0.12 to any port 5555 proto tcp

Si el tráfico sale de una red privada de confianza, activa el mecanismo de seguridad CURVE de ZeroMQ, que cifra y autentica cada conexión con pares de claves, o haz pasar el tráfico por una VPN como WireGuard.

Solución de problemas

  • zmq.error.ZMQError: Operation cannot be accomplished in current state: un socket REQ debe alternar estrictamente send y recv. Aparece si envías dos veces seguidas o si reutilizas el socket tras un tiempo de espera; en ese caso ciérralo y crea uno nuevo.
  • zmq.error.ZMQError: Address already in use: otro proceso ya hizo bind en ese puerto, a menudo una ejecución anterior que sigue viva. Búscala con sudo ss -tlnp | grep 5555.
  • El suscriptor no recibe nada: comprueba que el prefijo de subscribe() coincide exactamente con el inicio del mensaje, incluidas mayúsculas, y que no estás esperando mensajes publicados antes de conectarte.
  • El programa no termina al salir: por defecto, un contexto espera a enviar los mensajes pendientes. Configura zmq.LINGER a 0 en los sockets o ciérralos antes de context.term().

Conclusión

Has instalado ZeroMQ y pyzmq en Ubuntu 24.04 y programado los tres patrones fundamentales: petición-respuesta para llamadas síncronas, publicación-suscripción para difundir eventos y push-pull para repartir trabajo en paralelo. Como siguientes pasos, puedes combinar ROUTER y DEALER para construir un servidor asíncrono con varios trabajadores, activar CURVE para cifrar el tráfico entre máquinas o consultar la guía oficial de ZeroMQ (zguide.zeromq.org), que desarrolla patrones de fiabilidad más avanzados.