编辑:
如果使用 Python 3.8 或更高版本,则应使用asyncio
repl,如zeronone 的答案中所述。如果使用 3.7 或更低版本,您可以使用此答案。
您可以在后台线程中运行事件循环:
>>> import asyncio
>>>
>>> @asyncio.coroutine
... def greet_every_two_seconds():
... while True:
... print('Hello World')
... yield from asyncio.sleep(2)
...
>>> def loop_in_thread(loop):
... asyncio.set_event_loop(loop)
... loop.run_until_complete(greet_every_two_seconds())
...
>>>
>>> loop = asyncio.get_event_loop()
>>> import threading
>>> t = threading.Thread(target=loop_in_thread, args=(loop,))
>>> t.start()
Hello World
>>>
>>> Hello World
请注意,您必须调用asyncio.set_event_loop
,loop
否则您将收到一条错误消息,指出当前线程没有事件循环。
如果你想从主线程与事件循环交互,你需要坚持loop.call_soon_threadsafe
调用。
虽然这种事情是在解释器中进行试验的好方法,但在实际程序中,您可能希望所有代码都在事件循环中运行,而不是引入线程。