我是否以正确的方式管理 asyncio 任务（python 3.9）？

Question

我正在开发一个爬虫（python 3.9），我需要启动 make_attempt() 并根据其执行结果启动应该同时工作的其他任务。

首先我创建初始任务并将其添加到存储所有异步任务的列表中： self.data["worker"]["tasks"] 然后我启动： await asyncio.gather(*self.data["worker"]["tasks"])

在 make_attempt() 中，我等待向服务器发出 POST 请求的结果（我使用 aiohttp 客户端），然后根据结果，我要么添加新任务，要么在一小段延迟后重复 make_attempt()。

我停止当前任务并将其从异步任务列表中删除，然后添加新任务。

async def make_attempt(self):
    attempt: int = self.data["res"]["attempt"]

    await self.do_something()
    await sleep(1)

    for task in self.data["worker"]["tasks"]:
        print("Task name: %s" % task.get_name())
        if task.get_name() == str(attempt):
            task.cancel()
    self.data["worker"]["tasks"] = [task for task in self.data["worker"]["tasks"] if task.get_name() != str(attempt)]

    if 1 > 0:  # a condition to start make_attempt() again
        self.data["worker"]["tasks"].append(asyncio.create_task(self.make_attempt(), name=attempt))
        await asyncio.gather(*self.data["worker"]["tasks"])

async def run(self):
    self.data["worker"]["tasks"].append(asyncio.create_task(self.make_attempt(), name=self.data["res"]["attempt"]))
    await asyncio.gather(*self.data["worker"]["tasks"])

我是 asyncio 的新手，所以也许您可以指出错误或建议更好的实现。

更新。这是我想要实现的架构：

main_task 应该运行，如果结果正常，它应该启动另一个任务的几个实例（参见 loop2）。当得到 loop2 中任务的结果时，应该运行一个新的子任务。 main_task 应该等待 loop2 中的所有任务完成或触发 Timeout。 loop2 中的所有任务都应该同时工作。

UPD 2. 此代码在 check_base_url() 方法执行大约 1000 个周期后生成 RecursionError: maximum recursion depth exceeded while calling a Python object。

class Scraper:
    def __init__(self, data):
        self.data = data

    async def get_response(self, session, url, method="get", *args, **kwargs) -> Union[
        ClientResponse, None]:
        for _ in range(0, 20):
            if method == "get":
                try:
                    response = await session.get(url, headers={}, proxy="_proxy", *args, **kwargs)
                    if response.status > 399:
                        raise ScraperError(response.status)
                    await sleep(0.1)
                    return response
                except (ClientError, ScraperError) as err:
                    await sleep(0.25)
                    continue
            else:
                try:
                    response = await session.post(url, headers={}, proxy="_proxy", *args, **kwargs)
                    if response.status > 399:
                        raise ScraperError(response.status)
                    await sleep(0.1)
                    return response
                except (ClientError, ScraperError) as err:
                    await sleep(0.25)
                    continue
        return None

    async def get_captcha(self) -> SolvedCaptcha:
        for _ in range(0, 20):
            captcha = await self.task_1()
            if captcha:
                continue

    async def final_task(self, url) -> bool:
        async with ClientSession(cookies={}) as sess:
            resp_step1: Union[ClientResponse, None] = await self.get_response(sess, "url", "post",
                                                                              data={})
            if resp_step1:
                resp_step2: Union[ClientResponse, None] = await self.get_response(sess, "url", "get")
                if resp_step2:
                    captcha: SolvedCaptcha = await self.get_captcha()
                    if captcha:
                        resp_captcha: Union[ClientResponse, None] = await self.get_response(sess, "url",
                                                                                            "post",
                                                                                            data={})
                        if resp_captcha:
                            if 2 > 1:
                                print("FINISHED")
                                return True
                        else:
                            return False
                    else:
                        return False
                else:
                    return False
            else:
                return False

    async def add_task_3(self) -> None:
        if 2 > 1:
            subtasks = [asyncio.create_task(self.final_task(self.data["res"]["slots_urls"][0]))]
            await asyncio.gather(*subtasks)
        else:
            await self.add_task_3()

    def parse(self, html: str, url: str) -> None:
        soup = BeautifulSoup(html, "lxml")
        # do parsing

    async def task_2(self, url) -> bool:
        async with ClientSession() as sess:
            resp: Union[ClientResponse, None] = await self.get_response(sess, url)
            if not resp:
                return False
            html = await resp.text()
            self.parse(html, url)

    async def add_task_2(self) -> None:
        if 2 > 1:
            subtasks = [asyncio.create_task(self.task_2(url)) for url in ["url1", "url2"]]
            await asyncio.gather(*subtasks)

    async def task_1(self) -> bool:
        self.data["res"]["captcha_requested"] += 1
        res = await self.captcha.task_1()
        if not res:
            return False
        return True

    async def add_task_1(self) -> None:
        if 2 > 1:
            subtasks = [asyncio.create_task(self.task_1()) for _ in range(0, 5)]
            await asyncio.gather(*subtasks)

    async def get_calendar_url(self, sess) -> bool:
        resp: Union[ClientResponse, None] = await self.get_response(sess, "url", method="post",
                                                                    data={})
        if not resp:
            return False
        else:
            return True

    async def check_base_url(self) -> bool:
        async with ClientSession() as session_0:
            return await self.get_calendar_url(session_0)

    async def schedule_tasks(self):
        def start_again() -> bool:
            if 2 > 1:
                return True
            return False

        res_base_url: bool = await self.check_base_url()
        if res_base_url:
            tasks = [asyncio.create_task(self.add_task_1()),
                     asyncio.create_task(self.add_task_2()),
                     asyncio.create_task(self.add_task_3())]
            await asyncio.gather(*tasks)
            if start_again():
                await sleep(0.1)
                await self.schedule_tasks()
        else:
            await self.schedule_tasks()

    async def run(self):
        await self.schedule_tasks()

Answer 1

一种方法是使用asyncio.Queue 来管理应用程序中的待处理作业。然后，您可以创建多个从该队列接收作业的工作任务，也可以添加新作业。

在示例中，使用asyncio.sleep 模拟实际工作（POST 请求和处理一些数据）。有些工作会产生新的工作，而有些则不会。

工人将以这种方式同时接手和工作。

Manager 负责创建作业队列并等待所有项目被处理。之后它将取消所有工作人员并且程序终止。

代码

import asyncio
import random


class Worker:
    def __init__(self, num, target_q):
        # Worker number
        self.num = num
        # The job queue
        self.target_q = target_q
        # Create asyncio task
        self.task = asyncio.create_task(self.run())

    async def run(self):
        # Work on jobs until task is cancelled
        while True:
            print(f"Worker {self.num}: Waiting for new target")

            # Receive a new job from the queue
            target = await self.target_q.get()
            print(f"Worker {self.num}: Processing target {target}")

            try:
                # Simulating some work (e.g. a POST request)
                await asyncio.sleep(1.0)

                # Depending on the outcome, some new work results
                # 0-2 new targets are generated
                new_target_count = random.randint(0, 2)

                if new_target_count > 0:
                    print(
                        f"Worker {self.num}: Target {target} generating {new_target_count} more targets"
                    )
                    for _ in range(new_target_count):
                        # Create a new random target
                        new_target = random.randint(1, 10000)

                        # Put new targets into queue. This will wait if queue is
                        # currently full.
                        await self.target_q.put(new_target)

            finally:
                print(f"Worker {self.num}: Target {target} done")

                # Decrease queue count by one
                self.target_q.task_done()


class Manager:
    def __init__(self):
        # A common queue holding the jobs for the workers. It just stores
        # integers here but could hold any data.
        self.target_q = asyncio.Queue(10)
        # The list of workers
        self.workers = None

    async def run(self):
        # Create some initial work
        await self.target_q.put(1)

        # Create 3 workers
        self.workers = [Worker(num, self.target_q) for num in range(3)]

        # Wait until queue of unfinished tasks is empty
        await self.target_q.join()

        # Cancel other workers
        for worker in self.workers:
            worker.task.cancel()


def main():
    manager = Manager()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(manager.run())
    loop.close()


if __name__ == "__main__":
    main()

样本结果

$ python workers_test.py
Worker 0: Waiting for new target
Worker 0: Processing target 1
Worker 1: Waiting for new target
Worker 2: Waiting for new target
Worker 0: Target 1 generating 2 more targets
Worker 0: Target 1 done
Worker 0: Waiting for new target
Worker 0: Processing target 320
Worker 1: Processing target 5807
Worker 0: Target 320 done
Worker 0: Waiting for new target
Worker 1: Target 5807 done
Worker 1: Waiting for new target

Answer 2

asyncio.gather 在您添加新任务时不会更新。 您应该在需要的地方创建新任务。快速的经验法则你aio.gather()无论你在哪里做的一切都会并行执行。例如，总共将有 9 个some_async_task 并行运行

async def nested():
    tasks = [some_async_task(), some_async_task(), some_async_task()]
    await aio.gather(tasks)

async def main():
    tasks = [nested(), nested(), nested()]
    await aio.gather(tasks)

这是你可以做的：

import asyncio as aio

async def task():
    ...
    await subtask()

async def subtask():
    # anything here
    pass



async def main():
   r = await aio_http_request()
   if should_create_tasks(r):
       tasks = [task(), task(), task()]
       await aio.gather(tasks)
   else:
       await aio.sleep(5)
       await make_attempt()

 
async def run():
   await make_attempt()

这里 A、B、C 并发执行。而主任务 -> 任务 -> 子任务是顺序的。

+------------------------------------------------------------------------+  
|                                                                        |  
|                +-----------------+                                     |  
|                |                 |                                     |  
|                |    Main Task    |                                     |  
|                +-------+---------+                                     |  
|                        |                                               |  
|                        |                                               |  
|                        v                                               |  
|    +-+-------------+---+-------------++------+-------------+--------+  |  
|    | |      A      |   |      B      |       |      C      |        |  |  
|    | |  +-------+  |   |  +-------+  |       |  +-------+  |        |  |  
|    | |  |TASK   |  |   |  |TASK   |  |       |  |TASK   |  |        |  |  
|    | |  +---+---+  |   |  +---+---+  |       |  +---+---+  |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |  +---+---+  |   |  +---+---+  |       |  +---+---+  |        |  |  
|    | |  |subtask|  |   |  |subtask|  |       |  |subtask|  |        |  |  
|    | |  +-------+  |   |  +-------+  |       |  +-------+  |        |  |  
|    | |             |   |             |       |             |        |  |  
|    | +-------------+   +-------------+       +-------------+        |  |  
|    |                                                                |  |  
|    +----------------------------------------------------------------+  |  
|                                                                        |  
|                                                                        |  
|                                                                        |  
+------------------------------------------------------------------------+  

请记住，A 的 SUBTASK 可能在 B 的 TASK 完成之前发生，因为它们没有被绑定（它们是并发的）。

Answer 3

我不确定这就是你要找的东西?。 第一个。您不必存储coro。 第二。您在步骤create_task 上的代码无需等待即可将coro 抛出到async loop，但您再次调用await asyncio.gather。

如果您只想让它永远运行或退出if condition，请查看下面的代码。

async def make_attempt(self):
    await self.do_something()
    await asyncio.sleep(1)
    if 1 > 0:  # a condition to start make_attempt() again
        return await self.make_attempt()

async def run(self):
    await self.make_attempt()