How To Use Python pyzmq for Messaging with ZeroMQ

Intermediate

You have two Python processes that need to talk to each other. Maybe a data pipeline producer that scrapes data and a consumer that saves it to a database. Maybe a task queue where a coordinator dishes out jobs to a pool of workers. You try Python’s built-in multiprocessing.Queue and it works — until you need workers on different machines, or you want to add a third process in the middle, or the coordinator crashes and the whole system freezes. This is the wall that pyzmq helps you get past.

pyzmq is the Python binding for ZeroMQ, a high-performance asynchronous messaging library. ZeroMQ gives you sockets that handle connection management, message framing, retries, and backpressure — without requiring a central message broker like RabbitMQ or Redis. You install one package, import it, and you get production-grade inter-process communication that works across threads, processes, and machines on a network.

In this article we will cover what ZeroMQ is and how its socket types work, how to use the three most important messaging patterns — request/reply, push/pull, and publish/subscribe — how to handle multiple sockets at once with a poller, how to send structured data like JSON and Python objects, and how to build a real task-dispatching system. By the end you will have a solid grasp of pyzmq and know which pattern to reach for in your own projects.

pyzmq in Python: Quick Example

Here is a self-contained request/reply example — the most common ZeroMQ pattern. Run the server script first in one terminal, then the client in a second terminal:

# zmq_server.py
import zmq

context = zmq.Context()
socket = context.socket(zmq.REP)   # REP = reply socket
socket.bind("tcp://127.0.0.1:5555")

print("Server listening on port 5555...")
while True:
    message = socket.recv_string()
    print(f"Received: {message}")
    socket.send_string(f"Echo: {message}")

# zmq_client.py
import zmq

context = zmq.Context()
socket = context.socket(zmq.REQ)   # REQ = request socket
socket.connect("tcp://127.0.0.1:5555")

for i in range(3):
    socket.send_string(f"Hello {i}")
    reply = socket.recv_string()
    print(f"Reply: {reply}")

socket.close()
context.term()

Output (client terminal):

Reply: Echo: Hello 0
Reply: Echo: Hello 1
Reply: Echo: Hello 2

The key things happening here: context = zmq.Context() creates a ZeroMQ context — this manages all the sockets and background threads, and you only ever need one per process. The server calls socket.bind() while the client calls socket.connect() — this is the standard ZeroMQ convention. REP and REQ sockets are strictly paired and must alternate send/recv calls in lockstep.

The sections below go deeper into each pattern, including how to break out of the request/reply lockstep when you need more flexibility.

What Is ZeroMQ and Why Use It?

ZeroMQ (often written as 0MQ or ZMQ) is a messaging library — not a message broker. This is an important distinction. Tools like RabbitMQ or Kafka are brokers: they run as separate services that store and route messages. ZeroMQ is a library that gives your existing processes smart sockets. There is no broker to install, configure, or maintain. The sockets handle queuing, reconnection, and message framing internally.

Think of a regular TCP socket as a garden hose — reliable for moving water, but you have to manage the pressure, connections, and flow yourself. ZeroMQ sockets are more like a pneumatic tube system: you push an object in one end and it arrives intact at the other end, already framed, buffered, and ready to use. ZeroMQ also handles the case where the other end is temporarily unavailable — it queues messages and reconnects automatically.

Feature	Raw TCP / multiprocessing.Queue	ZeroMQ (pyzmq)
Message framing	Manual (you parse byte boundaries)	Automatic
Multiple patterns	Only point-to-point	REQ/REP, PUB/SUB, PUSH/PULL, DEALER/ROUTER
Cross-machine	Yes (TCP), No (Queue)	Yes (TCP, IPC, inproc)
Broker required	No	No
Backpressure	Manual	Built-in (HWM)
Fan-out / fan-in	Hard to implement	Native with right socket type

pyzmq is the official Python binding for ZeroMQ. It exposes the full ZeroMQ API and adds Python conveniences like send_string(), recv_json(), and async support via asyncio. Install it with pip install pyzmq — the package bundles the ZeroMQ C library so there is nothing extra to install.

The PUSH/PULL Pattern: Task Distribution

PUSH/PULL is the right pattern when you have one process producing work and multiple workers consuming it. The producer connects a PUSH socket and sends tasks; each worker connects a PULL socket and receives tasks. ZeroMQ distributes tasks round-robin across all connected workers automatically.

This example simulates a URL-checking pipeline. Run the producer in one terminal and one or more workers in separate terminals:

# push_producer.py
import zmq
import time

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

urls = [
    "https://httpbin.org/get",
    "https://httpbin.org/status/200",
    "https://httpbin.org/status/404",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/json",
]

print(f"Sending {len(urls)} tasks to workers...")
time.sleep(1)   # give workers time to connect

for url in urls:
    sender.send_string(url)
    print(f"Dispatched: {url}")

# Send poison pills to stop each worker
for _ in range(3):   # adjust to match worker count
    sender.send_string("STOP")

sender.close()
context.term()

# pull_worker.py
import zmq

context = zmq.Context()
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://127.0.0.1:5556")

print("Worker ready, waiting for tasks...")
while True:
    task = receiver.recv_string()
    if task == "STOP":
        print("Received stop signal, shutting down.")
        break
    print(f"Worker processing: {task}")

receiver.close()
context.term()

Output (one worker terminal):

Worker ready, waiting for tasks...
Worker processing: https://httpbin.org/get
Worker processing: https://httpbin.org/status/404
Worker processing: https://httpbin.org/json
Received stop signal, shutting down.

Notice that the producer calls bind() and the workers call connect() — the stable endpoint (the one that others connect to) should be the one that binds. The time.sleep(1) in the producer is a ZeroMQ gotcha: the first few milliseconds after connect() the socket is still handshaking, so messages sent immediately can be lost. A short sleep (or, better, a synchronization step) prevents the “slow joiner” problem.

The PUB/SUB Pattern: Broadcasting Events

PUB/SUB is for broadcast scenarios: one publisher sends messages and any number of subscribers receive them. Subscribers can filter by topic prefix, so a single publisher can serve many different audiences without sending irrelevant data to each one.

# pub_publisher.py
import zmq
import time
import random

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

topics = ["BTC", "ETH", "SOL"]
print("Publisher broadcasting prices...")
time.sleep(0.5)   # allow subscribers to connect

for _ in range(10):
    topic = random.choice(topics)
    price = round(random.uniform(1000, 60000), 2)
    message = f"{topic} {price}"
    publisher.send_string(message)
    print(f"Published: {message}")
    time.sleep(0.3)

publisher.close()
context.term()

# sub_subscriber.py
import zmq
import sys

topic_filter = sys.argv[1] if len(sys.argv) > 1 else "BTC"

context = zmq.Context()
subscriber = context.socket(zmq.SUB)
subscriber.connect("tcp://127.0.0.1:5557")
subscriber.setsockopt_string(zmq.SUBSCRIBE, topic_filter)

print(f"Subscribed to topic: {topic_filter}")
for _ in range(5):
    message = subscriber.recv_string()
    topic, price = message.split()
    print(f"Received -- {topic}: ${price}")

subscriber.close()
context.term()

Output (subscriber filtering for BTC):

Subscribed to topic: BTC
Received -- BTC: $42318.76
Received -- BTC: $41900.12
Received -- BTC: $43001.55
Received -- BTC: $42750.00
Received -- BTC: $41200.88

The critical line is subscriber.setsockopt_string(zmq.SUBSCRIBE, topic_filter) — without this, subscribers receive nothing. Setting the filter to an empty string "" subscribes to all messages. ZeroMQ filters are prefix-based: subscribing to "BTC" will match "BTC 42000", "BTCUSDT 42000", and anything else starting with those three characters. Design your topic strings accordingly.

Sending JSON and Python Objects

Real applications rarely send plain strings. pyzmq makes it easy to send JSON, bytes, or Python objects directly with a small set of convenience methods. Here is how to use them:

# zmq_json_example.py
import zmq
import json

# --- SERVER (in one process) ---
def run_server():
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind("tcp://127.0.0.1:5558")

    raw = socket.recv_json()           # deserializes JSON automatically
    print(f"Server received: {raw}")   # raw is already a Python dict

    response = {"status": "ok", "echo": raw["command"], "code": 200}
    socket.send_json(response)         # serializes dict to JSON automatically

    socket.close()
    context.term()

# --- CLIENT (in another process) ---
def run_client():
    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://127.0.0.1:5558")

    payload = {"command": "ping", "args": [1, 2, 3]}
    socket.send_json(payload)          # dict -> JSON -> bytes -> socket

    reply = socket.recv_json()         # bytes -> JSON -> dict
    print(f"Client received: {reply}")

    socket.close()
    context.term()

Output:

Server received: {'command': 'ping', 'args': [1, 2, 3]}
Client received: {'status': 'ok', 'echo': 'ping', 'code': 200}

send_json() and recv_json() handle serialization automatically — they call json.dumps() and json.loads() under the hood. For even better performance with large payloads, consider sending msgpack-encoded bytes via send() and recv(), since msgpack is faster and more compact than JSON for binary data.

Handling Multiple Sockets with a Poller

Sometimes a single process needs to listen on more than one socket at once — for example, a monitor that receives metrics from multiple sources. Blocking on one socket means you miss messages from the others. ZeroMQ’s Poller solves this: it watches multiple sockets simultaneously and returns whichever ones have messages ready.

# zmq_poller.py
import zmq

context = zmq.Context()

# Two sockets receiving data from different sources
metrics_socket = context.socket(zmq.PULL)
metrics_socket.bind("tcp://127.0.0.1:5559")

alerts_socket = context.socket(zmq.PULL)
alerts_socket.bind("tcp://127.0.0.1:5560")

poller = zmq.Poller()
poller.register(metrics_socket, zmq.POLLIN)
poller.register(alerts_socket, zmq.POLLIN)

print("Monitor listening on both sockets...")
message_count = 0
while message_count < 10:
    ready = dict(poller.poll(timeout=2000))   # wait up to 2 seconds

    if metrics_socket in ready:
        data = metrics_socket.recv_string()
        print(f"[METRIC]  {data}")
        message_count += 1

    if alerts_socket in ready:
        data = alerts_socket.recv_string()
        print(f"[ALERT]   {data}")
        message_count += 1

    if not ready:
        print("No messages received in 2 seconds -- still waiting.")

metrics_socket.close()
alerts_socket.close()
context.term()

Output (when fed by two separate sender scripts):

Monitor listening on both sockets...
[METRIC]  cpu=42% mem=61%
[ALERT]   Disk usage above 80% on /dev/sda1
[METRIC]  cpu=38% mem=59%
[METRIC]  cpu=51% mem=63%
[ALERT]   High latency detected: 320ms

poller.poll(timeout=2000) blocks for up to 2000 milliseconds and returns a dict mapping socket to event flags. If the returned dict is empty, the timeout elapsed with no activity. This pattern lets one process efficiently monitor many message sources without spinning in a busy loop or spawning threads for each socket.

Real-Life Example: Distributed Log Aggregator

Let us build a realistic system: multiple application processes push log messages to a central aggregator using the PUSH/PULL pattern. The aggregator writes them to a file and prints a running count every 5 messages.

# log_aggregator.py -- run this first
import zmq
import datetime

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

log_file = "aggregated.log"
count = 0

print(f"Log aggregator started. Writing to {log_file}")
with open(log_file, "a") as f:
    while True:
        try:
            entry = receiver.recv_json()
            timestamp = datetime.datetime.now().isoformat()
            line = f"[{timestamp}] [{entry['level']}] {entry['service']}: {entry['message']}\n"
            f.write(line)
            f.flush()
            count += 1
            if count % 5 == 0:
                print(f"Aggregated {count} log entries so far.")
        except KeyboardInterrupt:
            print(f"Shutting down. Total entries written: {count}")
            break

receiver.close()
context.term()

# log_sender.py -- run multiple instances of this
import zmq
import time
import random
import sys

service_name = sys.argv[1] if len(sys.argv) > 1 else "app"
levels = ["INFO", "WARNING", "ERROR"]
messages = [
    "Request processed in 45ms",
    "Cache miss on user profile",
    "Database connection pool at 80%",
    "Payment gateway timeout -- retrying",
    "Health check passed",
]

context = zmq.Context()
sender = context.socket(zmq.PUSH)
sender.connect("tcp://127.0.0.1:5561")

print(f"{service_name} sending logs...")
for _ in range(8):
    entry = {
        "service": service_name,
        "level": random.choice(levels),
        "message": random.choice(messages),
    }
    sender.send_json(entry)
    print(f"Sent: {entry['level']} -- {entry['message']}")
    time.sleep(random.uniform(0.2, 0.8))

sender.close()
context.term()

Aggregator output after two senders run:

Log aggregator started. Writing to aggregated.log
Aggregated 5 log entries so far.
Aggregated 10 log entries so far.
Aggregated 15 log entries so far.
^C
Shutting down. Total entries written: 16

This pattern scales horizontally: you can run 10 or 100 sender instances without changing the aggregator. ZeroMQ's PULL socket handles fair queuing automatically, so no single sender can flood the aggregator while others wait. You can extend this by adding a second PUSH/PULL stage that routes entries by log level to different downstream handlers, or by switching to PUB/SUB if you want multiple aggregators to each receive all entries.

Frequently Asked Questions

Does ZeroMQ need a broker like RabbitMQ?

No -- this is one of ZeroMQ's biggest selling points. ZeroMQ is a brokerless messaging library. The sockets themselves handle queuing, connection management, and message delivery without any separate server process. This makes deployment much simpler: you just install pyzmq, run your processes, and they talk to each other directly. The tradeoff is that there is no built-in message persistence or dead-letter queue -- if your consumer is down when a message is sent and the high-water mark is reached, messages can be dropped. If you need guaranteed delivery with persistence, consider adding a Redis or SQLite store alongside ZeroMQ.

What happens when a consumer is too slow?

ZeroMQ has a high-water mark (HWM) setting on each socket that limits the in-memory message queue. When the queue fills up (because a consumer is too slow), ZeroMQ's behavior depends on the socket type: PUSH sockets will block the sender until the consumer catches up, while PUB sockets will silently drop the oldest messages. You can tune the HWM with socket.set_hwm(1000) to control this. For production systems, monitor your queue depth and add more consumers before it approaches the HWM.

Are ZeroMQ sockets thread-safe?

No -- ZeroMQ sockets are not thread-safe. The rule is: one socket per thread, or use explicit locking. The zmq.Context() object is thread-safe and can be shared, but never pass a socket between threads without a lock. The idiomatic way to communicate between threads is to create a socket pair using the inproc:// transport, which avoids the network stack entirely and is extremely fast for same-process communication. You can also use zmq.devices.ThreadDevice to bridge sockets across threads safely.

Can I use pyzmq with asyncio?

Yes -- pyzmq has a dedicated zmq.asyncio module that provides async-compatible versions of the Context and Socket classes. You import it as import zmq.asyncio, create a context with zmq.asyncio.Context(), and then use await socket.recv_string() instead of socket.recv_string(). This integrates cleanly with asyncio.gather() and other async patterns, letting you handle ZeroMQ messages alongside HTTP requests, database calls, or other coroutines in a single event loop.

What are multipart messages and when do I use them?

ZeroMQ supports sending multiple message frames in one logical send operation using socket.send_multipart([frame1, frame2]) and receiving them with socket.recv_multipart(). This is used heavily by the DEALER/ROUTER pattern (a more advanced version of REQ/REP that adds routing envelopes) and is also useful when you want to send a message header separately from its body without serializing them into one structure. Each frame arrives atomically -- either all frames arrive or none do. For simple use cases, single-frame messages are simpler and sufficient.

Conclusion

We covered the three most important pyzmq patterns: request/reply (REQ/REP) for synchronous back-and-forth communication, push/pull (PUSH/PULL) for distributing work across multiple consumers, and publish/subscribe (PUB/SUB) for broadcasting events to many subscribers. We also saw how to send structured data with send_json()/recv_json(), how to monitor multiple sockets simultaneously with zmq.Poller, and how to build a real distributed log aggregator that scales horizontally without any broker.

The log aggregator example is a good starting point to extend. Try adding a second aggregation tier that routes ERROR entries to a separate file, or convert the PUSH/PULL senders to PUB sockets so you can tap into the log stream without modifying the senders. You could also swap the TCP transport for IPC (ipc:///tmp/logs.ipc) if all processes run on the same machine -- it is faster and avoids the network stack.

The full pyzmq documentation and a comprehensive guide to all ZeroMQ socket patterns are available at pyzmq.readthedocs.io and the excellent free book The ZeroMQ Guide (zguide.zeromq.org).

Related Articles

• How To Use Python celery-beat for Periodic Task Scheduling
• How To Use Python Prefect for Data Pipeline Orchestration
• How To Use Python socketio for Real-Time Web Events