python-3.x 如何以异步方式执行ZeroMQ PUSH/PULL原型？

6psbrbz9 于 2022-12-01 发布在 Python

关注(0)|答案(2)|浏览(80)

我想在一个端口上初始化一个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方法处停止。

python-3.x

来源：https://stackoverflow.com/questions/61275565/how-to-execute-a-zeromq-push-pull-archetype-in-an-asynchronous-way

2条答案

按热度按时间

m4pnthwp1#

#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()

赞(0）回复(0）举报 2022-12-01

fhg3lkii2#

上述代码未运行。*

代码正在运行，
然而在源代码中有错误（概念上）和打字错误。
代码的现状禁止**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号登月，或者相反，切尔诺贝利那种管理不善的制度+心态，是干净、鼓舞人心和警示性的例子，足以说明不良做法 （如果不加以禁止+发现+纠正+处罚） 会****一次又一次地危害**。

最佳下一步

如果你渴望达到专业水平，从Pieter Hintjens的书“Code Connected，Volume 1”开始--值得花时间，值得努力，值得理解他们讨论的概念。

赞(0）回复(0）举报 2022-12-01

我来回答

python-3.x 如何以异步方式执行ZeroMQ PUSH/PULL原型？

2条答案

溶液

最佳下一步

相关问题

热门标签

最新问答