Python asyncio：在不丢失状态的情况下从另一个任务停止和启动任务？

Question

我想从另一个任务中停止一个 python asyncio 任务，并在第二个任务中的某些条件发生时再次启动它。请注意，我不想取消第一个任务的协程（该协程停止时的状态应该可用）。另外，我不关心第一个任务所处的确切状态，我只希望事件循环停止运行第一个任务，直到第二个任务另行通知。我希望这个示例代码有助于理解这个问题：import asyncioasync def coroutine1():    i = 0    while(True):        i += 1        print("coroutine1: " + str(i) )        await asyncio.sleep(1)async def coroutine2(task1):    i = 0    while(True):        i += 1        if (i > 3) and (i<10):            pass #TODO: stop task1 here        else:            pass #TODO: Maybe check if task1 is running                 #and start task1 again if it's not?        print("coroutine2: " + str(i) )        await asyncio.sleep(1)async def main_coroutine():    loop = asyncio.get_event_loop()    task1 = loop.create_task(coroutine1())    task2 = loop.create_task(coroutine2(task1))    done, pending = await asyncio.wait(                [task1, task2]                , return_when=asyncio.FIRST_COMPLETED,)loop = asyncio.get_event_loop()loop.run_until_complete(main_coroutine())loop.close()

Answer 1

我想从另一个任务中停止一个 python asyncio 任务，并在第二个任务中的某些条件发生时再次启动它。

我假设您控制任务创建，但不想触及协程的实现。在您的情况下，您可以控制coroutine2和main_coroutine，但不能控制的内部coroutine1。

在这种情况下，您可以将协程包装在一个__await__that 中，而不是正常yield from循环，检查您的stopped标志并等待一个未来告诉它何时恢复。

class Stoppable:

    def __init__(self, coro):

        self._coro_iter = coro.__await__()

        self._stopped = None

    def __await__(self):

        while True:

            while self._stopped:

                print('awaiting stopped')

                yield from self._stopped.__await__()

            try:

                v = next(self._coro_iter)

            except StopIteration as e:

                return v

            yield v

    def stop(self):

        loop = asyncio.get_event_loop()

        self._stopped = loop.create_future()

    def start(self):

        if self._stopped is not None:

            self._stopped.set_result(None)

            self._stopped = None

您可以使用包装器进行修改coroutine2以随意停止和恢复执行coroutine1：

async def coroutine2(s):

    i = 0

    while True:

        i += 1

        if i == 3:

            print('stopping coroutine1')

            s.stop()

        elif i == 10:

            print('restarting coroutine1')

            s.start()

        print("coroutine2: " + str(i) )

        await asyncio.sleep(1)

async def main_coroutine():

    loop = asyncio.get_event_loop()

    s = Stoppable(coroutine1())

    fut1 = asyncio.ensure_future(s)

    task2 = loop.create_task(coroutine2(s))

    done, pending = await asyncio.wait(

        [fut1, task2], return_when=asyncio.FIRST_COMPLETED)

包装器的工作方式是展开yield from. 例如，要委托__await__给另一个协程，可以这样写：

    def __await__(self):

        yield from self._coro_iter

像这样编写，您无法实现停止，因为yield from包含一个隐式循环，该循环产生底层迭代器产生的所有值 - 类似于：

    def __await__(self):

        while True:

            try:

                v = next(self._coro_iter)

            except StopIteration as e:

                return e.value

            yield v

像这样，很容易在每次迭代时添加一个if检查_stopped，这意味着每次我们被事件循环恢复。剩下的障碍是在_stopped被取消之前不能只是忙循环——我们必须让出其他东西来允许事件循环继续运行其他协程。幸运的是，这很容易通过创造_stopped未来并从未来中实现。当设置了 future 的结果时，我们将自动恢复并继续执行包装的协程。