你的错误在于尝试在调用 app.run() 之后运行 asyncio 事件循环。后者不会返回，而是运行 Flask 开发服务器。
事实上，大多数 WSGI 设置都是这样工作的；要么主线程将忙于分派请求，要么 Flask 服务器作为模块在 WSGI 服务器中导入，在这里也无法启动事件循环。
相反，您需要在单独的线程中运行您的 asyncio 事件循环，然后通过asyncio.run_coroutine_threadsafe()在该单独线程中运行您的协程。有关此过程的详细信息，请参见文档中的协程和多线程部分。
这里是一个运行此类事件循环线程并提供您安排在该循环中运行协程的实用程序的模块实现：
import asyncio
import itertools
import threading
__all__ = ["EventLoopThread", "get_event_loop", "stop_event_loop", "run_coroutine"]
class EventLoopThread(threading.Thread):
loop = None
_count = itertools.count(0)
def __init__(self):
self.started = threading.Event()
name = f"{type(self).__name__}-{next(self._count)}"
super().__init__(name=name, daemon=True)
def __repr__(self):
loop, r, c, d = self.loop, False, True, False
if loop is not None:
r, c, d = loop.is_running(), loop.is_closed(), loop.get_debug()
return (
f"<{type(self).__name__} {self.name} id={self.ident} "
f"running={r} closed={c} debug={d}>"
)
def run(self):
self.loop = loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.call_later(0, self.started.set)
try:
loop.run_forever()
finally:
try:
shutdown_asyncgens = loop.shutdown_asyncgens()
except AttributeError:
pass
else:
loop.run_until_complete(shutdown_asyncgens)
try:
shutdown_executor = loop.shutdown_default_executor()
except AttributeError:
pass
else:
loop.run_until_complete(shutdown_executor)
asyncio.set_event_loop(None)
loop.close()
def stop(self):
loop, self.loop = self.loop, None
if loop is None:
return
loop.call_soon_threadsafe(loop.stop)
self.join()
_lock = threading.Lock()
_loop_thread = None
def get_event_loop():
global _loop_thread
if _loop_thread is None:
with _lock:
if _loop_thread is None:
_loop_thread = EventLoopThread()
_loop_thread.start()
_loop_thread.started.wait(1)
return _loop_thread.loop
def stop_event_loop():
global _loop_thread
with _lock:
if _loop_thread is not None:
_loop_thread.stop()
_loop_thread = None
def run_coroutine(coro):
"""Run the coroutine in the event loop running in a separate thread
Returns a Future, call Future.result() to get the output
"""
return asyncio.run_coroutine_threadsafe(coro, get_event_loop())
您可以使用此处定义的run_coroutine()函数来调度asyncio例程。使用返回的Future实例来控制协程：
- 使用
Future.result（）获取结果。您可以设置超时时间；如果在超时时间内没有产生结果，则自动取消协程。
- 您可以使用
.cancelled（）、.running（）和.done（）方法查询协程的状态。
- 您可以为Future添加回调函数，当协程完成、取消或引发异常时将调用该回调函数（请注意，这可能会从事件循环线程而不是您调用
run_coroutine（）的线程中调用）。
对于您的特定示例，其中abar()不返回任何结果，您可以忽略返回的future，就像这样：
@app.route("/")
def notify():
run_coroutine(abar("abar"))
return "OK"
请注意，在Python 3.8之前，您无法使用运行在单独线程上的事件循环来创建子进程！请参见我的回答 Python3 Flask asyncio subprocess in route hangs，了解Python 3.8 ThreadedChildWatcher类的后移解决方法。
app.run和loop.run_forever都是阻塞的。你最好使用线程。如果你必须使用 asyncio，那么你应该研究一下基于它构建的类似 Flask 的框架。 - dirn