TL;DR 要获取事件循环,请调用asyncio.get_event_loop()
.
在基于 asyncio 的应用程序中,事件循环通常不属于 Quart 或任何其他协议/应用程序级组件,它由 asyncio 或可能是像 uvloop 这样的加速器提供。事件循环是通过调用获得的asyncio.get_event_loop()
,有时设置为asyncio.set_event_loop()
.
这就是夸脱的app.run()
用于运行应用程序,这意味着它使用 asyncio 为主线程创建的默认事件循环。在你的情况下,你可以简单地调用夸脱run()
注册后Goblin
:
loop = asyncio.get_event_loop()
goblin_app = loop.run_until_complete(Goblin.open(loop))
goblin_app.register(Person, Knows)
quart_app = Quart(...)
# ... @app.route, etc
# now they both run in the same event loop
quart_app.run()
The above should answer the question in the practical sense. But that approach wouldn't work if more than one component insisted on having their own
run()
method that spins the event loop - since
app.run()
doesn't return, you can only invoke one such function in a thread.
但如果你仔细观察,就会发现事实并非如此quart
任何一个。虽然夸脱示例确实使用app.run()
为应用程序提供服务,如果你看一下实现app.run()
,你会看到它调用了便利函数run_app()
,它简单地创建了一个服务器并永远旋转主循环:
def run_app(...):
loop = asyncio.get_event_loop()
# ...
create_server = loop.create_server(
lambda: Server(app, loop, ...), host, port, ...)
server = loop.run_until_complete(create_server)
# ...
loop.run_forever()
如果您需要控制事件循环的实际运行方式,您始终可以自己完成:
# obtain the event loop from asyncio
loop = asyncio.get_event_loop()
# hook Goblin to the loop
goblin_app = loop.run_until_complete(Goblin.open(loop))
goblin_app.register(Person, Knows)
# hook Quart to the loop
quart_server = loop.run_until_complete(loop.create_server(
lambda: quart.serving.Server(quart_app, loop), host, port))
# actually run the loop (and the program)
try:
loop.run_forever()
except KeyboardInterrupt: # pragma: no cover
pass
finally:
quart_server.close()
loop.run_until_complete(quart_server.wait_closed())
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()