Example of push pull workers with zmq thread queue vs builtin thread queue

push_pull_threads_zmq.py

import time
import zmq
import zmq.devices
from threading import Thread

# Advantages:
#  - Easier to change the architecture around
#  - Easier to add other processes, workers, remote workers, etc
# Disadvantages:
#  - Need a little knowledge of zmq
#  - Need a package


def main():
    frontend_url = 'inproc://frontend'
    backend_url = 'inproc://backend'
    streamer = zmq.devices.ThreadDevice(zmq.STREAMER, zmq.PULL, zmq.PUSH)
    streamer.bind_in(frontend_url)
    streamer.bind_out(backend_url)
    streamer.start()
    threads = [
        Thread(target=pusher, args=(frontend_url, 0), daemon=True),
        Thread(target=pusher, args=(frontend_url, 1), daemon=True),
        Thread(target=puller, args=(backend_url, 2), daemon=True),
        Thread(target=puller, args=(backend_url, 3), daemon=True),
    ]
    for t in threads:
        t.start()
    while True:
        time.sleep(1)


def pusher(sock_url, _id):
    print(f'Pusher {_id} starting..')
    sock = zmq.Context.instance().socket(zmq.PUSH)
    sock.connect(sock_url)
    while True:
        sock.send_string('sup nerds')
        time.sleep(5)


def puller(sock_url, _id):
    print(f'Puller {_id} starting..')
    sock = zmq.Context.instance().socket(zmq.PULL)
    sock.connect(sock_url)
    while True:
        msg = sock.recv_string()
        print(f'Puller {_id} received: {msg}')


if __name__ == '__main__':
    main()