Python asyncio：如何在 Task.cancel 之后在线程之间安全地关闭事件循环？

Question

问题

在我自己的项目中，我在一个线程中启动一个事件循环，然后取消任务并安全地关闭另一个线程中的循环。但我失败了。

阅读task-object后， 我仍然无法理解如何在 Task.cancel 之后等待真正取消的任务

Python 版本：3.7.1

操作系统：windows

下面是我的调试过程。

import threading
import asyncio
import time

async def visit_sth():
    print("start sleep")
    await asyncio.sleep(3)
    print("end sleep")


class Person(object):

    def __init__(self):
        self.loop = asyncio.new_event_loop()

    def visit(self):

        asyncio.set_event_loop(self.loop)
        try:
            self.loop.run_until_complete(visit_sth())
        except Exception as err:
            print(err)

    def pause(self):

        tasks = asyncio.all_tasks(loop=self.loop)
        for t in tasks:
            t.cancel()

        self.loop.stop()
        self.loop.close()

P = Person()
t1 = threading.Thread(target=P.visit)
t2 = threading.Thread(target=P.pause)

t1.start()
time.sleep(0.5)
t2.start()

t1.join()
t2.join()

以下错误

start sleep
Exception in thread Thread-2:
Traceback (most recent call last):
  File "C:\Python3701\lib\threading.py", line 917, in _bootstrap_inner
    self.run()
  File "C:\Python3701\lib\threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "c:\Users\zhouxin\Documents\jupyterlab\learning_asyncio\starkoverflow.py", line 31, in pause
    self.loop.close()
  File "C:\Python3701\lib\asyncio\selector_events.py", line 94, in close
    raise RuntimeError("Cannot close a running event loop")
RuntimeError: Cannot close a running event loop

取消后，事件循环仍在运行。

另外，文档task-object 说Task.cancel() 不保证任务会被取消

所以我转向stackoverflow，阅读Kill tasks instead of awaiting them，然后更改pause方法，如

def pause(self):

    tasks = asyncio.all_tasks(loop=self.loop)
    for t in tasks:
        t.cancel()
        with suppress(asyncio.CancelledError):
            self.loop.run_until_complete(t)

    self.loop.stop()
    self.loop.close()

另一个错误发生了

start sleep
Exception in thread Thread-2:
Traceback (most recent call last):
  File "C:\Python3701\lib\threading.py", line 917, in _bootstrap_inner
    self.run()
  File "C:\Python3701\lib\threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "c:\Users\zhouxin\Documents\jupyterlab\learning_asyncio\starkoverflow.py", line 31, in pause
    self.loop.run_until_complete(t)
  File "C:\Python3701\lib\asyncio\base_events.py", line 560, in run_until_complete
    self.run_forever()
  File "C:\Python3701\lib\asyncio\base_events.py", line 515, in run_forever
    raise RuntimeError('This event loop is already running')
RuntimeError: This event loop is already running

这种方法行不通。

现在，我真的很困惑如何等待任务取消然后在 Task.cancel 之后关闭循环。

Answer 1

您的代码有几个问题：

• pause 从运行事件循环的线程外部与事件循环交互。这是禁止的，必须替换为对run_coroutine_threadsafe 和call_soon_threadsafe 的调用。

• 代码为每个业务对象创建一个新的事件循环。这是不可取的，因为强大的 asyncio 套件允许在单个事件循环中使用多个协程。

推荐的模式是有一个单一的事件循环，并使用run_coroutine_threadsafe 向它提交任务。当你想取消一个任务时，而不是停止整个循环，你只需取消那个特定的任务。例如：

import threading
import asyncio
import time

async def visit_sth():
    print("start sleep")
    await asyncio.sleep(3)
    print("end sleep")


class Person:
    def visit(self):
        # returns a concurrent.futures.Future
        self._visit_fut = asyncio.run_coroutine_threadsafe(visit_sth(), loop)
        # result() waits for the future to be resolved and returns the
        # result
        try:
            return self._visit_fut.result()
        except Exception as err:
            print(type(err), err)

    def pause(self):
        self._visit_fut.cancel()

loop = asyncio.new_event_loop()
threading.Thread(target=loop.run_forever).start()

P = Person()
t1 = threading.Thread(target=P.visit)
t2 = threading.Thread(target=P.pause)

t1.start()
time.sleep(0.5)
t2.start()

t1.join()
t2.join()
loop.call_soon_threadsafe(loop.stop)