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

Question

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

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

如果每个 MiniBot 在其他 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 只是将控制权交还给事件循环，并在所有提供的任务/ future 完成时返回完成的。您可以将其替换为类似 await asyncio.sleep(10) 的内容，并具有相同的效果(在此示例中)。而且，如果您有不可预测数量的 future ，还有其他其他方式来表示结束条件。

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