等待异步协程按特定顺序完成

Question

我有一个包含一堆 asyncio.coroutine 的守护程序，可以总结为类似这样的内容

import asyncio
import signal

class Daemon:

    def __init__(self, loop=asyncio.get_event_loop()):
        self.loop = loop
        self.running = False
        self.tasks = {
            'coroutine1': asyncio.ensure_future(self.coroutine1()),
            'coroutine2': asyncio.ensure_future(self.coroutine2()),
        }

    def run(self):
        self.running = True
        for task in self.tasks.values():
            task.add_done_callback(self.task_done_callback)
        # gracefuly close everything when SIGINT (could be ^C) is received
        self.loop.add_signal_handler(signal.SIGINT, self.close)
        self.loop.run_forever()

    def close(self):
        self.running = False
        self.loop.run_until_complete(self.tasks['coroutine1'])
        self.loop.run_until_complete(self.tasks['coroutine2'])

    def task_done_callback(self, future):
        for task in self.tasks.values():
            if not task.done():
                return
        self.loop.stop()

    @asyncio.coroutine
    def coroutine1(self):
        while self.running:
            print('coroutine1: do stuff')
            yield from asyncio.sleep(1)

    @asyncio.coroutine
    def coroutine2(self):
        while self.running:
            print('coroutine2: do some other stuff')
            yield from asyncio.sleep(3)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    daemon = Daemon(loop)
    daemon.run()
    loop.close()

当程序接收到SIGINT 时，守护程序会正常关闭。当这种情况发生时，close() 方法被触发，该方法负责通知所有正在运行的任务它们应该完成工作并停止。这只需将running 设置为False 即可。每当一个任务完成时，Daemon.task_done_callback 就会被触发。它检查是否所有任务都已完成，如果是，则停止循环。

这里的问题是close() 方法不起作用。那是因为我在循环已经运行时调用了loop.run_until_complete（通过run_forever）。这将产生一个RuntimeError('This event loop is already running')。

重要的是：coroutine1 需要在 coroutine2 之前完成，因为如果 coroutine2 不再做它的事情，coroutine1 可能会遇到问题。

我的问题是我如何确保coroutine1 在coroutine2 之前完成？

Answer 1

这是实现这一目标的一种方式。 （我删除了signal 部分，只是限制了coroutine1 的运行时间）。

import asyncio

class Daemon:

    def __init__(self, loop=asyncio.get_event_loop()):
        self.loop = loop
        self.tasks = {
            'coroutine1': asyncio.ensure_future(self.coroutine1()),
            'coroutine2': asyncio.ensure_future(self.coroutine2())}

    def run(self):
        for task in self.tasks.values():
            task.add_done_callback(self.task_done_callback)
        self.loop.run_forever()

    def task_done_callback(self, future):
        if all(task.done() for task in self.tasks.values()):
            self.loop.stop()

    @asyncio.coroutine
    def coroutine1(self):
        for _ in range(5):
            print('coroutine1: doing stuff')
            yield from asyncio.sleep(0.2)
        print('coroutine1: done!')

    @asyncio.coroutine
    def coroutine2(self):
        while not self.tasks['coroutine1'].done():
            print('coroutine2: doing stuff while coro1 is running')
            yield from asyncio.sleep(0.2)
        print('coroutine2: doing stuff after coro1 has ended')
        yield from asyncio.sleep(1)
        print('coroutine2: done!')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    daemon = Daemon(loop)
    daemon.run()

这里主要是检查coroutine1 是否仍在运行（您的self.tasks 属性可以被查询）。

现在将其与signal 集成以停止coroutine1 我建议您注册一个设置标志的简单函数（例如self.signal_flag）。然后在coroutine1 中循环使用while not self.signal_flag: ... 之类的东西。这些是我建议的完整解决方案的更改：

class Daemon:

    def __init__(self, loop=asyncio.get_event_loop()):
        ...
        self.signal_flag = False

    def run(self):
        ...
        self.loop.add_signal_handler(signal.SIGINT, self.set_signal_flag)
        self.loop.run_forever()

    def set_signal_flag(self):
        print('caught signal!')
        self.signal_flag = True

    async def coroutine1(self):
        while not self.signal_flag:
            print('coroutine1: doing stuff')
            await asyncio.sleep(0.2)
        print('coroutine1: done!')

还请注意，不需要close 循环。在解释器继续运行的 IDE 中，这会使事情变得比他们需要的更复杂......

从 python 3.5 开始，您可以（并且可能应该）为您的协程使用这种语法：

async def coroutine1(self):
    for _ in range(5):
        print('coroutine1: doing stuff')
        await asyncio.sleep(0.2)
    print('coroutine1: done!')