上面的代码有几个问题:
-
stop()
不需要参数
- 程序在协程执行之前结束(
stop()
在它之前被调用)。
这是固定代码(没有异常和异常处理程序):
import asyncio
from threading import Thread
async def coro():
print("in coro")
return 42
loop = asyncio.get_event_loop()
thread = Thread(target=loop.run_forever)
thread.start()
fut = asyncio.run_coroutine_threadsafe(coro(), loop)
print(fut.result())
loop.call_soon_threadsafe(loop.stop)
thread.join()
call_soon_threadsafe()
返回一个保存异常的 future 对象(它不会到达默认的异常处理程序):
import asyncio
from pprint import pprint
from threading import Thread
def exception_handler(loop, context):
print('Exception handler called')
pprint(context)
loop = asyncio.get_event_loop()
loop.set_exception_handler(exception_handler)
thread = Thread(target=loop.run_forever)
thread.start()
async def coro():
print("coro")
raise RuntimeError("BOOM!")
fut = asyncio.run_coroutine_threadsafe(coro(), loop)
try:
print("success:", fut.result())
except:
print("exception:", fut.exception())
loop.call_soon_threadsafe(loop.stop)
thread.join()
然而,使用调用的协程create_task()
or ensure_future()
将调用异常处理程序:
async def coro2():
print("coro2")
raise RuntimeError("BOOM2!")
async def coro():
loop.create_task(coro2())
print("coro")
raise RuntimeError("BOOM!")
您可以使用它来创建一个小包装器:
async def boom(x):
print("boom", x)
raise RuntimeError("BOOM!")
async def call_later(coro, *args, **kwargs):
loop.create_task(coro(*args, **kwargs))
return "ok"
fut = asyncio.run_coroutine_threadsafe(call_later(boom, 7), loop)
但是,您可能应该考虑使用Queue https://docs.python.org/3/library/queue.html与您的线程进行通信。