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

Question

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

请注意，我不想取消第一个任务的协程（该协程停止时的状态应该可用）。另外，我不关心第一个任务所处的确切状态，我只希望事件循环停止运行第一个任务，直到第二个任务另行通知。

我希望这个示例代码有助于理解这个问题：

import asyncio

async 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:&nbsp; &nbsp; def __init__(self, coro):&nbsp; &nbsp; &nbsp; &nbsp; self._coro_iter = coro.__await__()&nbsp; &nbsp; &nbsp; &nbsp; self._stopped = None&nbsp; &nbsp; def __await__(self):&nbsp; &nbsp; &nbsp; &nbsp; while True:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; while self._stopped:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print('awaiting stopped')&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; yield from self._stopped.__await__()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; v = next(self._coro_iter)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; except StopIteration as e:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return v&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; yield v&nbsp; &nbsp; def stop(self):&nbsp; &nbsp; &nbsp; &nbsp; loop = asyncio.get_event_loop()&nbsp; &nbsp; &nbsp; &nbsp; self._stopped = loop.create_future()&nbsp; &nbsp; def start(self):&nbsp; &nbsp; &nbsp; &nbsp; if self._stopped is not None:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self._stopped.set_result(None)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self._stopped = None您可以使用包装器进行修改coroutine2以随意停止和恢复执行coroutine1：async def coroutine2(s):&nbsp; &nbsp; i = 0&nbsp; &nbsp; while True:&nbsp; &nbsp; &nbsp; &nbsp; i += 1&nbsp; &nbsp; &nbsp; &nbsp; if i == 3:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print('stopping coroutine1')&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; s.stop()&nbsp; &nbsp; &nbsp; &nbsp; elif i == 10:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print('restarting coroutine1')&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; s.start()&nbsp; &nbsp; &nbsp; &nbsp; print("coroutine2: " + str(i) )&nbsp; &nbsp; &nbsp; &nbsp; await asyncio.sleep(1)async def main_coroutine():&nbsp; &nbsp; loop = asyncio.get_event_loop()&nbsp; &nbsp; s = Stoppable(coroutine1())&nbsp; &nbsp; fut1 = asyncio.ensure_future(s)&nbsp; &nbsp; task2 = loop.create_task(coroutine2(s))&nbsp; &nbsp; done, pending = await asyncio.wait(&nbsp; &nbsp; &nbsp; &nbsp; [fut1, task2], return_when=asyncio.FIRST_COMPLETED)包装器的工作方式是展开yield from. 例如，要委托__await__给另一个协程，可以这样写：&nbsp; &nbsp; def __await__(self):&nbsp; &nbsp; &nbsp; &nbsp; yield from self._coro_iter像这样编写，您无法实现停止，因为yield from包含一个隐式循环，该循环产生底层迭代器产生的所有值 - 类似于：&nbsp; &nbsp; def __await__(self):&nbsp; &nbsp; &nbsp; &nbsp; while True:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; try:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; v = next(self._coro_iter)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; except StopIteration as e:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return e.value&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; yield v像这样，很容易在每次迭代时添加一个if检查_stopped，这意味着每次我们被事件循环恢复。剩下的障碍是在_stopped被取消之前不能只是忙循环——我们必须让出其他东西来允许事件循环继续运行其他协程。幸运的是，这很容易通过创造_stopped未来并从未来中实现。当设置了 future 的结果时，我们将自动恢复并继续执行包装的协程。