python asyncio，如何从另一个线程创建和取消任务

Question

我有一个python多线程应用程序。我想在一个线程中运行一个异步循环，并从另一个线程向它发送回调和协程。应该很容易，但是我不能把我的头放在那些东西上。

我提出了以下解决方案，其中一半是我想要的，可以随意评论任何事情：

import asyncio
from threading import Thread

class B(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.loop = None

    def run(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop) #why do I need that??
        self.loop.run_forever()

    def stop(self):
        self.loop.call_soon_threadsafe(self.loop.stop)

    def add_task(self, coro):
        """this method should return a task object, that I
          can cancel, not a handle"""
        f = functools.partial(self.loop.create_task, coro)
        return self.loop.call_soon_threadsafe(f)

    def cancel_task(self, xx):
        #no idea

@asyncio.coroutine
def test():
    while True:
        print("running")
        yield from asyncio.sleep(1)

b.start()
time.sleep(1) #need to wait for loop to start
t = b.add_task(test())
time.sleep(10)
#here the program runs fine but how can I cancel the task?

b.stop()

所以开始和停止循环都很好。我考虑过使用create_task创建任务，但该方法不是线程安全的，所以我将其包装在call_soon_threadsafe中。但我希望能够获取任务对象以便能够取消任务。我可以用未来和条件做一件复杂的事情，但必须有一个更简单的方法，不是吗？

Answer 1

只是作为参考，我最终实现的代码基于我在这个网站上得到的帮助，它更简单，因为我不需要所有的功能。再次感谢！

import asyncio
from threading import Thread
from concurrent.futures import Future
import functools

class B(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.loop = None

    def run(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    def stop(self):
        self.loop.call_soon_threadsafe(self.loop.stop)

    def _add_task(self, future, coro):
        task = self.loop.create_task(coro)
        future.set_result(task)

    def add_task(self, coro):
        future = Future()
        p = functools.partial(self._add_task, future, coro)
        self.loop.call_soon_threadsafe(p)
        return future.result() #block until result is available

    def cancel(self, task):
        self.loop.call_soon_threadsafe(task.cancel)

Answer 2

你什么都做对了。 用于任务停止生成方法

class B(Thread):
    # ...
    def cancel(self, task):
        self.loop.call_soon_threadsafe(task.cancel)

顺便说一下，可以通过

self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)

因为asyncio只为主线程创建隐式事件循环。

Answer 3

我认为您可能需要让您的add_task方法知道它是否从事件循环以外的线程调用。这样，如果它是从同一线程调用的，您可以直接调用asyncio.async，否则，它可以做一些额外的工作将任务从循环线程传递到调用线程。下面是一个例子：

import time
import asyncio
import functools
from threading import Thread, current_thread, Event
from concurrent.futures import Future

class B(Thread):
    def __init__(self, start_event):
        Thread.__init__(self)
        self.loop = None
        self.tid = None
        self.event = start_event

    def run(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        self.tid = current_thread()
        self.loop.call_soon(self.event.set)
        self.loop.run_forever()

    def stop(self):
        self.loop.call_soon_threadsafe(self.loop.stop)

    def add_task(self, coro):
        """this method should return a task object, that I
          can cancel, not a handle"""
        def _async_add(func, fut):
            try:
                ret = func()
                fut.set_result(ret)
            except Exception as e:
                fut.set_exception(e)

        f = functools.partial(asyncio.async, coro, loop=self.loop)
        if current_thread() == self.tid:
            return f() # We can call directly if we're not going between threads.
        else:
            # We're in a non-event loop thread so we use a Future
            # to get the task from the event loop thread once
            # it's ready.
            fut = Future()
            self.loop.call_soon_threadsafe(_async_add, f, fut)
            return fut.result()

    def cancel_task(self, task):
        self.loop.call_soon_threadsafe(task.cancel)


@asyncio.coroutine
def test():
    while True:
        print("running")
        yield from asyncio.sleep(1)

event = Event()
b = B(event)
b.start()
event.wait() # Let the loop's thread signal us, rather than sleeping
t = b.add_task(test()) # This is a real task
time.sleep(10)
b.stop()

首先，我们将事件循环的线程id保存在run方法中，这样我们就可以确定稍后对add_task的调用是否来自其他线程。如果add_task是从非事件循环线程调用的，我们使用call_soon_threadsafe调用一个同时调度协程的函数，然后使用concurrent.futures.Future将任务传递回调用线程，该线程等待Future的结果。

关于取消任务的说明：当您在Task上调用cancel时，下次运行事件循环时，将在协程中引发CancelledError。这意味着，任务正在包装的协程将在下次到达屈服点时由于异常而中止，除非协程捕获CancelledError，并阻止其自身中止。还要注意，这只在被包装的函数实际上是一个可中断的协程的情况下才有效；例如，由BaseEventLoop.run_in_executor返回的asyncio.Future实际上不能被取消，因为它实际上被包装在concurrent.futures.Future上，并且一旦它们的底层函数真正开始执行，这些函数就不能被取消。在这些情况下，asyncio.Future会说它被取消了，但是实际在执行器中运行的函数将继续运行。

编辑：根据Andrew Svetlov的建议，更新了第一个示例以使用concurrent.futures.Future，而不是queue.Queue。

注意：由于版本3.4.4使用^{}，因此不推荐使用^{}。