The most surprising thing about using ZeroMQ with Tornado is how little ZeroMQ actually needs to know about Tornado’s event loop to integrate effectively.
Let’s see ZeroMQ in action with Tornado. Imagine a simple publisher/subscriber setup. Our publisher will send out messages every second, and our subscriber will receive and print them.
First, the publisher. It’s a standard ZeroMQ PUB socket, but we’ll bind it to a local address.
import zmq
import time
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:5556")
print("Publisher started on tcp://127.0.0.1:5556")
topic = "updates"
message_count = 0
while True:
message = f"{topic} {message_count}".encode('utf-8')
socket.send(message)
print(f"Sent: {message.decode('utf-8')}")
message_count += 1
time.sleep(1)
Now, the subscriber. This is where Tornado comes in. We need to integrate the ZeroMQ socket into Tornado’s IOLoop. ZeroMQ sockets, unlike regular Python sockets, don’t directly expose the file descriptors that Tornado’s IOLoop typically watches. However, pyzmq provides a way to get an object that IOLoop can work with.
import zmq
from tornado.ioloop import IOLoop
from tornado.tcpserver import TCPServer
from tornado.iostream import Stream
class ZMQSubscriber:
def __init__(self, io_loop: IOLoop):
self.io_loop = io_loop
self.context = zmq.Context()
self.socket = self.context.socket(zmq.SUB)
self.socket.setsockopt_string(zmq.SUBSCRIBE, "updates")
self.socket.connect("tcp://127.0.0.1:5556")
print("Subscriber connected to tcp://127.0.0.1:5556")
# Get the file descriptor for the ZeroMQ socket
self.zmq_fd = self.socket.getsockopt(zmq.FD)
# Register the file descriptor with Tornado's IOLoop for read events
self.io_loop.add_handler(self.zmq_fd, self.handle_message, io_loop.READ)
def handle_message(self, fd, events):
# This callback will be triggered by Tornado when the ZeroMQ socket is readable
try:
message = self.socket.recv_string()
print(f"Received: {message}")
except zmq.Again:
# This can happen if the event loop fires but there's no message yet
pass
except Exception as e:
print(f"Error receiving message: {e}")
def stop(self):
self.io_loop.remove_handler(self.zmq_fd)
self.socket.close()
self.context.term()
print("Subscriber stopped")
if __name__ == "__main__":
io_loop = IOLoop.current()
subscriber = ZMQSubscriber(io_loop)
print("Starting Tornado IOLoop...")
try:
io_loop.start()
except KeyboardInterrupt:
print("Stopping IOLoop...")
subscriber.stop()
io_loop.stop()
print("Stopped.")
When you run the publisher and subscriber scripts, you’ll see the subscriber receiving messages published by the publisher, all managed by Tornado’s event loop. The key here is self.socket.getsockopt(zmq.FD) which gives us the underlying file descriptor that Tornado can monitor. self.io_loop.add_handler() then tells Tornado to call self.handle_message whenever that file descriptor has data ready to be read.
The problem this solves is enabling asynchronous I/O for ZeroMQ within a Tornado application. Traditionally, if you wanted to use ZeroMQ in a web server like Tornado, you’d either block the event loop with synchronous ZeroMQ calls or manage separate threads. This integration allows ZeroMQ operations to be non-blocking and participate directly in Tornado’s event-driven architecture.
Internally, pyzmq leverages ZeroMQ’s pollable nature. ZeroMQ sockets can be "polled" to see if they are ready for reading or writing without actually performing the read/write operation. Tornado’s IOLoop does exactly this: it monitors a set of file descriptors and wakes up the application when one of them becomes ready. By registering the ZeroMQ socket’s file descriptor with Tornado, we’re essentially saying, "Hey Tornado, let me know when this ZeroMQ socket has data."
The exact lever you control is how you register the socket’s file descriptor with the IOLoop. You can add handlers for READ, WRITE, and ERROR events. For a subscriber, READ is primary. For a publisher that might be sending large messages and wants to know when it can send more without blocking, WRITE would be relevant. The events argument passed to your handler function tells you why the handler was called (e.g., io_loop.READ means the socket is ready for reading).
The zmq.FD option is not a standard ZeroMQ option across all language bindings; it’s specific to how pyzmq exposes ZeroMQ’s underlying mechanisms and how Tornado expects file descriptors. This makes the pyzmq library the crucial bridge.
If you were to build a more complex system with multiple ZeroMQ sockets and Tornado handlers, you’d notice that IOLoop.add_callback is essential for ensuring that any work triggered by ZeroMQ events (like processing a message) happens within the Tornado event loop’s context, preventing potential race conditions or deadlocks if you were to directly call blocking operations from within the handle_message function.
The next problem you’ll likely encounter is managing the lifecycle of your ZeroMQ context and sockets, especially when dealing with application shutdown or dynamic reconfigurations, ensuring all resources are properly cleaned up to avoid leaks.