如何以异步方式执行ZeroMQ PUSH / PULL原型？

Question

根据我的要求，我想在端口中启动一个PULL，并希望从其他端口接收到我的PULL端口。对于PULL端口，它异步侦听，并且在收到消息时，仅在控制台中打印该消息。为此，我在Push类中编写了一个方法，该方法会将消息发送到PULL端口。我的代码如下：

import random
import zmq
import time
import sys
import string
import asyncio
import zmq.asyncio

class Push():
    def __init__(self, port, addr='localhost'):
        self.port = port
        self.addr = addr

        self.ctx = zmq.Context()
        self.scoket = self.ctx.socket(zmq.PUSH)
        self.scoket.connect(f'tcp://{self.addr}:{selfa.port}')

    def send(self):
        chars = string.ascii_uppercase + string.ascii_lowercase
        message = ''.join(random.choice(chars) for _ in range(4))
        self.scoket.send(bytes(message, 'utf-8'))
        print(f'sending: {message}')

class Pull():
    def __init__(self, port, addr='*'):
        self.port = port
        self.addr = addr

        self.ctx = zmq.Context()
        self.socket = self.ctx.socket(zmq.PULL)
        self.socket.bind(f'tcp://{self.addr}:{self.port}')

    async def listen(self, listener):
        while True:
            string = await self.socket.recv()
            listener(string)


if __name__ == '__main__':
    push = Push('55501')

    async def send():
        while True:
            await asyncio.sleep(5)
            print('Sending...')
            push.send()

    pull = Pull('55501')

    try:
        asyncio.run(
            pull.listen(print),
            send(),
        )
    except KeyboardInterrupt:
        print('exiting...')
        exit()

以上代码未运行。该代码在listen方法处停止。请帮忙 ！！

Answer 1

#ADAPTED FROM PYMATA EXPRESS EXAMPLE CONCURRENTTAKS
#https://github.com/MrYsLab/pymata-express/
import asyncio
import zmq
import json
import zmq.asyncio as zmq_asyncio
from pymata_express.pymata_express import PymataExpress


class ConcurrentTasks:

    def __init__(self, board):


        self.loop = board.get_event_loop()
        self.board = board

        self.ctxsync = zmq.Context()
        self.context = zmq.asyncio.Context()
        self.rep = self.context.socket(zmq.REP)
        self.rep.bind("tcp://*:5558")

        self.trigger_pin = 53
        self.echo_pin = 51

        loop.run_until_complete(self.async_init_and_run())

    ### START:  NEW CODE THAT RESOLVED THE ISSUE
    async def pingsonar(self):
        value = await self.board.sonar_read(self.trigger_pin)
        return value

    async def readsonar(self):
        while True:
            rep_recv = await self.rep.recv() 
            value = await asyncio.wait([self.pingsonar()])
            valuesonar = list(value[0])[0].result()
            json_data = json.dumps(valuesonar) 
            await self.rep.send(json_data.encode()) 
            await asyncio.sleep(1 / 1000) #maybe this line isn't necessary

    ### END : NEW CODE THAT RESOLVED THE ISSUE

    async def async_init_and_run(self):

        await self.board.set_pin_mode_sonar(self.trigger_pin, self.echo_pin)

        readsonar = asyncio.create_task(self.readsonar())
        await readsonar

        # OTHER CREATED_TASK GO HERE, (removed them in the MVE, but they work fine)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    my_board = PymataExpress()
    try:
        ConcurrentTasks(my_board)
    except (KeyboardInterrupt, RuntimeError):
        loop.run_until_complete(my_board.shutdown())
        print('goodbye')
    finally:
        loop.close()

Answer 2

Q：上面的代码未运行。 ... 请帮助！

代码正在运行，但是源代码中存在错误（概念上）和错别字。

代码原样禁止Push-类成为.connect()-ed，因此Pull交易对手.bind() -s，却没有人要交谈。

---

解决方案

1]修复此错字（+更好，显式检测并处理所有潜在的错误状态）

self.scoket.connect(f'tcp://{self.addr}:{selfa.port}') # this will NEVER FLY
#--------------------------------------------^
self.scoket.connect(f'tcp://{self.addr}:{self.port}')  # this will ... ( + detect Error-state(s)

2]纠正概念-混合低挂的成果（一次使用几个异步框架，一个对所有其他框架进行工作）表明对专业工程在可靠且足够稳固的责任方面的理解不足distributed-computing（控制概念上的缺陷，如干净，启发和警告足够的例子，例如在阿波罗11号登月或相反的切尔诺贝利式管理不善+心态上的缺陷。不良做法（如果未被禁止+被发现+纠正+受到处罚）可能会＆将会危害一次又一次地）。

---

_{如果从未使用过ZeroMQ，在这里您可以先看看"ZeroMQ Principles in less than Five Seconds"，然后再深入研究更多细节]]}

---

最佳下一步

[如果确实渴望达到专业水平，请从Pieter HINTJENS的书“ Code Connected，第1卷开始-值得花时间，值得努力，值得理解那里讨论的概念。