使用 asyncio 转换一个简单的多线程程序

Question

我对 Python asyncio 还是很陌生，所以我试图将我使用多线程解决的一个简单问题转换为使用 asyncio。

我举了一个例子来说明我想要实现的目标。 每个 MiniBot 实例都可以在随机时间启动（main 中的 time.sleep() 调用表示在不可预测时间的实例化。）

如果每个 MiniBot 在其他人完成之前启动，我希望它们能够并行运行。

多线程很好，但是当我用异步协程翻译问题时，我无法让它们一起开始。

我可以使用gather，但这需要在开始时完成所有任务，而我没有。 有什么建议吗？

谢谢

哦，是的，我使用的是 Python 3.6

多线程版本

import threading
import time

class MiniBot(object):
    def __init__(self, _id:str):
        self._id = _id
        self.alive = True
        self.start_time = time.time()
        self.th = threading.Thread(target=self.run)

        self.internal_state = 0


    def _foo(self):
        self.internal_state += 1


    def run(self):
        while self.alive:
            self._foo()

            if time.time() - self.start_time > 4:
                print(f"Killing minibot: {self._id}")
                print(f"Var is: {self.internal_state}")
                self.stop()
            time.sleep(0.1)


    def start(self):
        print(f"Starting Minibot {self._id}")
        self.th.start()


    def stop(self):
        self.alive = False


if __name__ == "__main__":
    # MiniBots activities start at random times but should coexist
    MiniBot('a').start()
    time.sleep(2)
    MiniBot('b').start()
    time.sleep(1.5)
    MiniBot('c').start()

输出：

Starting Minibot a
Starting Minibot b
Starting Minibot c
Killing minibot: a
Var is: 40
Killing minibot: b
Var is: 40
Killing minibot: c
Var is: 40

异步版本（表现不如我所愿）

import asyncio
import time

class MiniBot(object):
    def __init__(self, _id:str):
        self._id = _id
        self.alive = True
        self.last_event = time.time()

        self.internal_state = 0


    async def _foo(self):
        self.internal_state += 1
        asyncio.sleep(2)


    async def run(self):
        while self.alive:
            await self._foo()

            if time.time() - self.last_event > 4:
                print(f"Killing minibot: {self._id}")
                print(f"Var is: {self.internal_state}")
                self.stop()
            asyncio.sleep(0.1)


    def start(self):
        print(f"Starting Minibot {self._id}")
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            loop.run_until_complete(self.run())
        finally:
            loop.run_until_complete(loop.shutdown_asyncgens())
            loop.close()


    def stop(self):
        self.alive = False


if __name__ == "__main__":
    # MiniBots activities start at random times but should coexist
    MiniBot('a').start()
    time.sleep(2)
    MiniBot('b').start()
    time.sleep(1.5)
    MiniBot('c').start()

输出：

Starting Minibot a
Killing minibot: a
Var is: 2839119
Starting Minibot b
Killing minibot: b
Var is: 2820634
Starting Minibot c
Killing minibot: c
Var is: 2804579

Answer 1

start 不能调用run_until_complete 因为run_until_complete 运行一个协程到最后，而你需要多个协程并行运行。创建和启动线程的 asyncio 等效项是 asyncio.create_task()，因此 start() 应该调用它，然后返回给调用者，就像在线程版本中一样。

例如：

import asyncio, time

class MiniBot:
    def __init__(self, _id):
        self._id = _id
        self.alive = True
        self.start_time = time.time()
        self.internal_state = 0

    def _foo(self):
        self.internal_state += 1

    async def run(self):
        while self.alive:
            self._foo()
            if time.time() - self.start_time > 4:
                print(f"Killing minibot: {self._id}")
                print(f"Var is: {self.internal_state}")
                self.stop()
            await asyncio.sleep(0.1)

    def start(self):
        print(f"Starting Minibot {self._id}")
        return asyncio.create_task(self.run())

    def stop(self):
        self.alive = False

async def main():
    taska = MiniBot('a').start()
    await asyncio.sleep(2)
    taskb = MiniBot('b').start()
    await asyncio.sleep(1.5)
    taskc = MiniBot('c').start()
    await asyncio.gather(taska, taskb, taskc)

if __name__ == "__main__":
    #asyncio.run(main())
    asyncio.get_event_loop().run_until_complete(main())

不要让对gather() 的调用让您失望：gather 只是将控制权交还给事件循环，并在所有提供的任务/未来完成后返回。您可以将其替换为 await asyncio.sleep(10) 之类的内容，并具有相同的效果（在此示例中）。如果你有不可预测的未来数量，还有其他方法可以表示结束条件。

另外注意，需要等待asyncio.sleep()，否则无效。