如何在 Python 中运行 class 级别的异步任务？

Question

我想启动一个基于 classed 的异步 task/thread。类似于 class 变量，这意味着 class.

的每个实例都可以访问它

我尝试创建一个使用 asyncio 的单独 class。

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)

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

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

Answer 1

如果你想从每个 asyncio coroutine/task 修改 list_of_tick，你必须使它成为 class 级别字段：

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 class 级别字段，我会把它放在 class 之外。但这更符合我的口味，两种方式都对。

编辑：

我想在另一个 class 中使用测试任务，您可以按以下方式执行：

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())

如何在 Python 中运行 class 级别的异步任务？

How to run an asynchronous task on class level in Python?

python

asynchronous

class

python-asyncio

如何在 Python 中 运行 class 级别的异步任务？

How to run an asynchronous task on class level in Python?

python

asynchronous

class

python-asyncio

如何在 Python 中运行 class 级别的异步任务？