永远在线程内运行 Python 3 asyncio.coroutine答案

【问题标题】：Running a Python 3 asyncio.coroutine inside a thread forever永远在线程内运行 Python 3 asyncio.coroutine
【发布时间】：2020-03-17 03:09:07
【问题描述】：

在尝试在线程中运行 HBMQTT 消息代理时，我对 asyncio 有点陌生。手册给出了如何启动代理的以下示例：

import asyncio
import os
from hbmqtt.broker import Broker

@asyncio.coroutine
def broker_coro():
    broker = Broker()
    yield from broker.start()

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(broker_coro())
    asyncio.get_event_loop().run_forever()

由于我所依赖的架构，代理必须在线程内运行。不幸的是，以下基本示例在线程调用 run() 之前崩溃。

import asyncio
from threading import Thread
from hbmqtt.broker import Broker

class ExampleThread(Thread):
    def __init__(self):
        super().__init__()
        self.daemon = True
        self.config = {
            'listeners': {
                'default': {
                    'max-connections': 50000,
                    'bind': 'localhost:1883',
                    'type': 'tcp',
                },
            },
            'auth': {
                'allow-anonymous': True,
            },
            'plugins': [ 'auth_anonymous' ],
            'topic-check': {
                'enabled': False
            }
        }
        self.loop = None
        self.broker = None

    @asyncio.coroutine
    def broker_coroutine(self):
        self.broker = Broker(self.config, self.loop)
        yield from self.broker.start()
        return self.broker

    def run(self) -> None:
        print('running ...')
        self.loop.run_forever()
        self.loop.run_until_complete(self.broker.shutdown())
        self.loop.close()

    def start(self):
        print('starting thread ...')
        self.loop = asyncio.new_event_loop()
        print('starting server ...')
        try:
            start_server = asyncio.gather(self.broker_coroutine(),
                                          loop=self.loop)
            self.loop.run_until_complete(start_server)
            broker = start_server.result()[0]
        except:
            print(traceback.format_exc())
            self.loop.close()

        super().start()


if __name__ == '__main__':
    thread = ExampleThread()
    thread.start()

启动示例会引发以下异常：

$ python3.7 ./mqtt.py
starting thread ...
starting server ...
Task was destroyed but it is pending!
task: <Task pending coro=<Broker._broadcast_loop() running at venv/lib/python3.7/site-packages/hbmqtt/broker.py:696> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x8024e7350>()]>>
Exception ignored in: <generator object Broker._broadcast_loop at 0x8027d1e50>
Traceback (most recent call last):
  File "venv/lib/python3.7/site-packages/hbmqtt/broker.py", line 696, in _broadcast_loop
  File "/usr/local/lib/python3.7/asyncio/queues.py", line 161, in get
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 687, in call_soon
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 479, in _check_closed
RuntimeError: Event loop is closed

谁能解释导致事件循环关闭的原因？如果我运行一个简单的测试协程，它可以工作：

async def test_coroutine(self):
    while True:
        await asyncio.sleep(1)
        print('hey!')

【问题讨论】：

您是否已采取措施调试问题？协程根本不运行，还是不在单独的线程中运行？你有例外吗？您是否尝试过添加打印以查看该程序单独运行的程度？请编辑问题以包含此附加信息。
代理似乎没有收听任何传入的消息。我用一个基本的协程替换了协程，它只在循环内运行yield from asyncio.sleep(delay)，并有一些输出到stdout，但没有打印任何内容。
new_event_loop 没有为当前线程设置事件循环。我认为这就是你的协程没有运行的原因。您只处理 KeyboardInterrupts 但抑制其他异常，因此如果对 _run_until_complete 的调用失败，您将看不到任何回溯。我绝对不确定这一点，因为关于这一点的文档根本不清楚。

标签： python python-3.x python-asyncio

【解决方案1】：

我无法在 Thread 类中运行 asyncio.coroutine，而是在一个杰出的 Thread 中运行，如下例所示：

import asyncio
from threading import Thread
from hbmqtt.broker import Broker

class Server():

    def __init__(self):
        self.broker = None
        self.config = {
            'listeners': {
                'default': {
                    'max-connections': 50000,
                    'bind': '127.0.0.1:1883',
                    'type': 'tcp',
                },
            },
            'plugins': [ 'auth_anonymous' ],
            'auth': {
                'allow-anonymous': True,
            },
            'topic-check': {
                'enabled': True,
                'plugins': ['topic_taboo'],
            },
        }

    async def broker_coroutine(self, config, loop):
        self.broker = Broker(config, loop)
        await self.broker.start()

    def start(self):
        loop = asyncio.new_event_loop()
        thread = Thread(target=lambda: self.run(loop))
        thread.start()

    def run(self, loop):
        try:
            future = asyncio.gather(self.broker_coroutine(self.config, loop),
                                    loop=loop,
                                    return_exceptions=True)
            loop.run_until_complete(future)
            loop.run_forever()
        except (Exception, KeyboardInterrupt):
            loop.close()
        finally:
            loop.run_until_complete(self.broker.shutdown())
            loop.close()

if __name__ == '__main__':
    server = Server()
    server.start()

【讨论】：