Python websockets、asyncio、队列 - 自定义服务器类和具有生产者和消费者方法的处理程序类

Question

我在理解 asyncio 工作流时遇到了问题......

编辑 - 更改代码以合并 asyncio.Queue:

#!/usr/bin/env python

import asyncio
import websockets
import threading


class WSServer:

    def serve_forever(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        start_server = websockets.serve(self.handler, '127.0.0.1', 5678)
        asyncio.get_event_loop().run_until_complete(start_server)
        asyncio.get_event_loop().run_forever()

    async def handler(self, websocket, path):
        loop = asyncio.get_event_loop()
        master = MyClass(websocket)
        while True:
            listener_task = asyncio.ensure_future(master.get_message())
            producer_task = asyncio.ensure_future(master.produce())
            done, pending = await asyncio.wait(
                [listener_task, producer_task],
                return_when=asyncio.FIRST_COMPLETED)

            if listener_task in done:
                await master.consume()
            else:
                listener_task.cancel()

            if producer_task in done:
                msg_to_send = producer_task.result()
                await master.send_message(msg_to_send)
            else:
                producer_task.cancel()


class MyClass:

    incoming = asyncio.Queue()
    outgoing = asyncio.Queue()

    def __init__(self, websocket):
        self.ws = websocket

    async def get_message(self):
        msg_in = await self.ws.recv()
        await self.incoming.put(msg_in)

    async def send_message(self, message):
        await self.ws.send(message)

    async def consume(self):
        msg_to_consume = await self.incoming.get()
        # do something 'consuming' :)
        consume_output = msg_to_consume
        await self.outgoing.put(consume_output)

    async def produce(self):
        msg_out = await self.outgoing.get()
        return msg_out


if __name__ == '__main__':
    s = WSServer()
    t = threading.Thread(target=s.serve_forever)
    t.daemon = True
    t.start()
    while True:
        asyncio.sleep(5)

当改变 MyClass.consume() 时它可以工作(在一台机器上，在另一台机器上不行哈哈)，但有奇怪的行为:

async def consume(self):
    msg_to_consume = await self.incoming.get()
    # do something 'consuming' :)
    consume_output = msg_to_consume
    await self.outgoing.put('THIS WILL NOT GET INTO QUEUE???!!!')
    print('Outgoing empty 1: ' + str(self.outgoing.empty()))
    # And this will get into queue O.o
    await self.outgoing.put(consume_output)
    print('Outgoing empty 2: ' + str(self.outgoing.empty()))

我有两个等待，因为在第一次调用 self.outgoing.put() 之后，self.outgoing 队列仍然是空的!只有当我再次调用它时，它才似乎收到了元素……有什么想法吗？

其他机器只是抛出错误:

Exception in connection handler
Traceback (most recent call last):
  File "/usr/lib/python3/dist-packages/websockets/server.py", line 78, in handler
    yield from self.ws_handler(self, path)
  File "test2.py", line 33, in handler
    msg_to_send = producer_task.result()
  File "/usr/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/usr/lib/python3.5/asyncio/tasks.py", line 241, in _step
    result = coro.throw(exc)
  File "test2.py", line 66, in produce
    msg_out = await self.outgoing.get()
  File "/usr/lib/python3.5/asyncio/queues.py", line 168, in get
    yield from getter
  File "/usr/lib/python3.5/asyncio/futures.py", line 361, in __iter__
    yield self  # This tells Task to wait for completion.
RuntimeError: Task <Task pending coro=<MyClass.produce() running at test2.py:66> cb=[_wait.<locals>._on_completion() at /usr/lib/python3.5/asyncio/tasks.py:414]> got Future <Future pending> attached to a different loop

原创:

我有这段代码，但显然没有按我的预期工作:)

#!/usr/bin/env python

import asyncio
import websockets
import threading


class WSServer:

    def serve_forever(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        start_server = websockets.serve(self.handler, '127.0.0.1', 5678)
        asyncio.get_event_loop().run_until_complete(start_server)
        asyncio.get_event_loop().run_forever()

    async def handler(self, websocket, path):
        loop = asyncio.get_event_loop()
        master = MyClass(websocket)
        while True:
            listener_task = asyncio.ensure_future(master.get_message())
            producer_task = asyncio.ensure_future(master.produce())
            done, pending = await asyncio.wait(
                [listener_task, producer_task],
                return_when=asyncio.FIRST_COMPLETED)

            if listener_task in done:
                await master.consume()
            else:
                listener_task.cancel()

            if producer_task in done:
                if producer_task.result():
                    await master.send_message()
            else:
                producer_task.cancel()


class MyClass:

    incoming = []
    outgoing = []

    def __init__(self, websocket):
        self.ws = websocket

    async def get_message(self):
        self.incoming.append(self.ws.recv())

    async def send_message(self):
        self.ws.send(self.outgoing.pop(0))

    async def consume(self):
        self.outgoing.append(self.incoming.pop(0))

    async def produce(self):
        if self.outgoing:
            return True


if __name__ == '__main__':
    s = WSServer()
    t = threading.Thread(target=s.serve_forever)
    t.daemon = True
    t.start()
    while True:
        asyncio.sleep(5)

我要实现的目标:

1. 让 WSServer 实例在与主线程不同的线程中运行(与 WSServer.serve_forever 一起正常工作)

2. 对于每个连接的客户端，在 WSServer.handler 方法中创建包含两个列表的 MyClass 实例 - 一个用于传入消息，另一个用于传出。

3. 传入应该从 MyClass.get_message() 填充 - 基本上是 websocket.recv()

4. 传出可以从 MyClass.consume() 填充 - 作为响应，但也可以从该代码范围之外填充。

5. 当MyClass.incoming有内容时，通过MyClass.consume()处理，当Myclass.outgoing中有内容时，通过MyClass.send_message()处理

我不确定 MyClass.produce()，因为我真的不需要生成任何东西，只需在有消息时发送消息即可。我也看到了一些使用 asycnio.Queue() 的代码

我在这里找到了类似的线程，但老实说，它们的示例和问题超出了我的理解范围:

Need help on producer and consumer thread in python

asyncio queue consumer coroutine

这里正确的方法应该是什么？

Answer 1

在python聊天室的帮助下找到了答案。

class MyClass:

    def __init__(self, websocket):
        self.ws = websocket
        self.incoming = asyncio.Queue()
        self.outgoing = asyncio.Queue()

队列应该为类的实例定义，而不是类本身。