时间戳

首页 > 解决方案 > 带有 aiozmq 流的简单 PUB/SUB

python - 带有 aiozmq 流的简单 PUB/SUB

问题描述

我正在尝试使用 aiozmq 流制作一个简单的 PUB/SUB（由于某些原因我不想使用 aiozmq rpc）但没有成功：

发布.py

# coding: utf-8
import asyncio
import time

import aiozmq
import zmq


async def do():
    stream = await aiozmq.stream.create_zmq_stream(
        zmq_type=zmq.PUB,
        bind='tcp://127.0.0.1:5556',
    )
    while True:
        await asyncio.sleep(1)
        msg = [str(time.time()).encode()]
        print('write ', msg)
        stream.write(msg)

loop = asyncio.get_event_loop()
loop.run_until_complete(do())

子.py

# coding: utf-8
import asyncio

import aiozmq
import zmq


async def do():
    stream = await aiozmq.stream.create_zmq_stream(
        zmq_type=zmq.SUB,
        connect='tcp://127.0.0.1:5556',
    )
    while True:
        print('wait ...')
        msg = await stream.read()
        print('received ', msg)

loop = asyncio.get_event_loop()
loop.run_until_complete(do())

执行 pub.py 时：

python pub.py      
write  [b'1534927086.914483']
write  [b'1534927087.9154818']
write  [b'1534927088.9164672']

然后执行 sub.py：

python sub.py
wait ...

我错过了什么？

标签： pythonzeromq

解决方案

只需错过sub.py. 有一个工作 sub.py：

# coding: utf-8
import asyncio

import aiozmq
import zmq


async def do():
    stream = await aiozmq.stream.create_zmq_stream(
        zmq_type=zmq.SUB,
        connect='tcp://127.0.0.1:5556',
    )
    stream.transport.subscribe(b'')

    while True:
        print('wait ...')
        msg = await stream.read()
        print('received ', msg)

loop = asyncio.get_event_loop()
loop.run_until_complete(do())

谁生产：

python sub.py 
wait ...
received  [b'1534927504.0462704']
wait ...
received  [b'1534927505.0478334']

推荐阅读