带有 asyncio 和 aiohttp 的异步程序的最佳架构

Question

我正在尝试了解如何最好地构建执行以下操作的程序：

考虑多重分析。每个分析都从多个数据源（REST API）请求数据。在每次分析中，当从数据源收集所有数据时，会检查数据的一个或多个条件。如果满足这些条件，则会向另一个数据源发出另一个请求。

目标是异步收集所有分析的数据，检查每个分析的条件，请求是否满足条件，然后重复。因此，以下要求：

1. 在在特定分析中收集所有数据之后检查数据的条件，而不是在所有分析中收集数据之后。
2. 如果满足条件，则首先提出请求 - 而不是在检查所有分析的条件之后。
3. 获取数据 -> 检查条件 -> 可能请求某些内容，循环安排为每 X 分钟或几小时运行一次。

我做了以下演示：

import asyncio
import random


async def get_data(list_of_data_calls):
    tasks = []
    for l in list_of_data_calls:
        tasks.append(asyncio.ensure_future(custom_sleep(l)))
    return await asyncio.gather(*tasks)


async def custom_sleep(time):
    await asyncio.sleep(time)
    return random.randint(0, 100)


async def analysis1_wrapper():
    while True:
        print("Getting data for analysis 1")
        res = await get_data([5, 3])
        print("Data collected for analysis 1")
        for integer in res:
            if integer > 80:
                print("Condition analysis 1 met")
            else:
                print("Condition analysis 1 not met")
        await asyncio.sleep(10)


async def analysis2_wrapper():
    while True:
        print("Getting data for analysis 2")
        res = await get_data([5, 3])
        print("Data collected for analysis 2")
        for integer in res:
            if integer > 50:
                print("Condition analysis 2 met")
            else:
                print("Condition analysis 2 not met")
        await asyncio.sleep(10)


loop = asyncio.get_event_loop()
tasks = analysis1_wrapper(), analysis2_wrapper()
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()

这会产生以下输出：

Getting data for analysis 2
Getting data for analysis 1
Data collected for analysis 2
Condition analysis 2 not met
Condition analysis 2 not met
Data collected for analysis 1
Condition analysis 1 not met
Condition analysis 1 not met
Getting data for analysis 2
Getting data for analysis 1
Data collected for analysis 2
Condition analysis 2 met
Condition analysis 2 not met
Data collected for analysis 1
Condition analysis 1 not met
Condition analysis 1 not met
Getting data for analysis 2
Getting data for analysis 1
Data collected for analysis 2
Condition analysis 2 not met
Condition analysis 2 not met
Data collected for analysis 1
Condition analysis 1 not met
Condition analysis 1 not met

这似乎可以按我的意愿工作。但是，由于我对 asyncio 和 aiohttp 的经验有限，我不确定这是否是一个好方法。我希望将来能够向管道添加步骤，例如如果满足条件，则根据发出的请求的逻辑执行某些操作。此外，它应该可以扩展到许多分析而不会损失太多速度。

Answer 1

是的，基本上就是这样。需要考虑的几点：

1. 并发限制。

虽然你可能有无限数量的并发任务，但随着并发的增加，每个任务的响应时间也会增加，吞吐量在某个时候停止增长甚至下降。因为只有一个主线程来执行所有操作，所以当回调太多时，即使网络响应在几毫秒前到达，回调也必须排队。为了平衡这一点，您通常需要一个Semaphore 来控制最大并发性能。

1. CPU 密集型操作。

您的代码没有显示，但令人担心的是条件检查可能会占用大量 CPU。在这种情况下，您应该将任务推迟到thread pool（没有 GIL 问题）或子进程（如果 GIL 是一个问题），原因有二： 1. 阻止阻塞的主线程损害并发性。 2. 更有效地利用多个 CPU。

1. 任务控制。

您当前的代码在每次分析时循环休眠 10 秒。这使得优雅地关闭分析仪变得很困难，更不用说动态缩放了。一个理想的模型是producer-consumer 模式，在这种模式下，您可以在Queue 中生成具有某种控制的任务，然后一群工作人员从队列中检索任务并同时处理它们。