时间戳

首页 > 解决方案 > 如何在 Python 中的类级别上运行异步任务？

python - 如何在 Python 中的类级别上运行异步任务？

问题描述

我想启动一个基于分类的异步任务/线程。就像类变量一样，这意味着类的每个实例都可以访问它。

我试图创建一个使用 asyncio 的单独类。

class yahoo_connection:

    def __init__(self):
        self.list_of_tick=[]
        self.format="%H   %M   %S"
        self.running=True
        asyncio.create_task(self.const_thread())

    async def const_thread(self):
        while self.running==True:
            print("current list: ",self.list_of_tick)
            if len(self.list_of_tick)!=0:
                idx=np.random.randint(len(self.list_of_tick)+1)
                self.format=self.list_of_tick[idx]
            await asyncio.sleep(3.5)

所以这是我计划初始化为我的原始类的类线程的类。但是，此代码不起作用。

我希望能够在每个实例中修改 self.list_of_tick 使用该类yahoo_connection作为类异步线程的类。

标签： pythonclassasynchronouspython-asyncio

解决方案

如果要从每个 asyncio 协程/任务中修改 list_of_tick，则必须将其设为类级别字段：

import asyncio


class Test:
    storage = []  # class level field

    def __init__(self, name):
        self.name = name  # instance level field

    async def do_work(self):
        """just add some digits to class level field"""
        for i in range(5):
            Test.storage.append(i)
            print(f"{self.name}: {Test.storage}")
            await asyncio.sleep(1)


async def async_main():
    """create three class instances and run do_work"""
    await asyncio.gather(*(Test(f"Cor-{i}").do_work() for i in range(3)))

if __name__ == '__main__':
    asyncio.run(async_main())

输出：

Cor-0: [0] Cor-1: [0, 0] Cor-2: [0, 0, 0] Cor-0: [0, 0, 0, 1] Cor-1: [0, 0, 0 , 1, 1] Cor-2: [0, 0, 0, 1, 1, 1] Cor-0: [0, 0, 0, 1, 1, 1, 2] Cor-1: [0, 0, 0, 1, 1, 1, 2, 2] Cor-2: [0, 0, 0, 1, 1, 1, 2, 2, 2] Cor-0: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3] Cor-1: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3] Cor-2: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3, 3] Cor-0: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 4] Cor-1 : [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4] Cor-2: [0, 0, 0, 1, 1, 1, 2, 2 , 2, 3, 3, 3, 4, 4, 4]

但如果我是你的，我不会做 list_of_tick 类级别字段，我只会把它放在课堂之外。但更多的是我自己的口味，两种方式都是对的。

编辑：

我想在另一个类中使用 Test 任务，您可以通过以下方式进行操作：

import asyncio


class Test:
    storage = []  # class level field

    def __init__(self, name):
        self.name = name  # instance level field

    async def do_work(self):
        """just add some digits to class level field"""
        for i in range(3):
            Test.storage.append(i)
            print(f"{self.name}: {Test.storage}")
            await asyncio.sleep(1)


class TestWrapper:
    def __init__(self, n):
        self.n = n

    async def run(self):
        await asyncio.gather(*(Test(f"Cor-{i}").do_work() for i in range(self.n)))


if __name__ == '__main__':
    asyncio.run(TestWrapper(3).run())

推荐阅读