gistfile1.py

import asyncio
import zmq
import aiozmq.interface


class TProt(aiozmq.interface.ZmqProtocol):
    def __init__(self):
        self.queue = asyncio.Queue()
        self.server = False

    def connection_made(self, transport):
        self.transp = transport
        super().connection_made(transport)

    def msg_received(self, data):
        self.transp.write(data)
        if self.server:
            counter.col += 1


@asyncio.coroutine
def counter():
    while True:
        yield from asyncio.sleep(1)
        print(counter.col)
        counter.col = 0

counter.col = 0


@asyncio.coroutine
def server():
    loop = asyncio.get_event_loop()
    transp, proto = yield from loop.create_zmq_connection(lambda: TProt(), zmq.ROUTER, bind='tcp://127.0.0.1:5555')
    proto.server = True


@asyncio.coroutine
def client():
    loop = asyncio.get_event_loop()
    transp, proto = yield from loop.create_zmq_connection(
        lambda: TProt(), zmq.DEALER, connect='tcp://127.0.0.1:5555')

    transp.write((b'Hello',))


asyncio.set_event_loop_policy(aiozmq.ZmqEventLoopPolicy())
asyncio.async(counter())
asyncio.async(server())
asyncio.async(client())
asyncio.get_event_loop().run_forever()
#asyncio.get_event_loop().run_until_complete(client())

gistfile2.py

# 5500 rps

import threading
import zmq
import time


def counter():
    while True:
        time.sleep(1)
        col, counter.col = counter.col, 0
        print(col)

counter.col = 0


def client():
    time.sleep(0.3)

    context = zmq.Context()
    cl = context.socket(zmq.DEALER)

    cl.connect("tcp://localhost:5555")

    msg = b'hello'
    cl.send_multipart([msg])
    while True:
        msg = cl.recv_multipart()
        cl.send_multipart(msg)

threading.Thread(target=counter).start()
threading.Thread(target=client).start()

context = zmq.Context()
srv = context.socket(zmq.ROUTER)

srv.bind("tcp://*:5555")

while True:
    #print(repr(srv.recv()))
    data = srv.recv_multipart()
    #print(data)
    counter.col += 1
    srv.send_multipart(data)