asyncio/aiohttp - 如何发出一系列异步但相关的请求？

Question

我有一系列异步请求对，其中一对由请求 A 和请求 B 组成。此外，请求 B 依赖于请求 A。换句话说，我需要将数据从响应 A 传递到请求 B。因此，我需要安排任务，使得每个任务发送请求 A，然后仅在响应 A 返回后发送请求 B。

from aiohttp import ClientSession
from typing import *
import asyncio

async def request_A(url: str, session: ClientSession) -> dict:
    async with session.request('get', url) as response:
        return await response.json()

async def request_B(url: str, data: dict, session: ClientSession) -> dict:
    async with session.request('post', url, json=data) as response:
        return await response.json()

async def request_chain(url_A: str, url_B: str, session: ClientSession) -> dict:
    response_A_data = await request_A(url_A, session)
    response_B_data = await request_B(url_B, response_A_data, session)
    return response_B_data

async def schedule(url_chains: List[Tuple[str, str]]) -> list:
    tasks = []
    async with ClientSession() as session:
        for url_chain in url_chains:
            url_A, url_B = url_chain
            task = asyncio.create_task(request_chain(url_A, url_B, session))
            tasks.append(task)
        return await asyncio.gather(*tasks)

def run_tasks(url_chains: List[Tuple[str, str]]) -> list:
    return asyncio.run(schedule(url_chains))

现在，我的问题是：对于由一对请求组成的每个任务，请求 A 是否能保证在发送请求 B 之前返回？请解释。我担心在任务中，当请求 A 正在等待时，请求 B 可能会执行。

如果不是，我如何保持任务异步和非阻塞，同时确保在任务内，请求 A 阻止请求 B 的执行，直到响应 A 返回？

据我所知，我可以批量运行所有请求 A 调用，然后批量运行所有请求 B 调用，但由于我的用例特定的原因，我需要运行一批所有（请求 A，请求 B）对。

Answer 1

对于由一对请求组成的每个任务，是否保证请求 A 在发送请求 B 之前返回？

是的，async/await 模式的优点是您不必问自己这个问题，连续的代码行将始终按顺序执行（但不一定连续）。这里你的函数request_chain保证request_A总是会在 之前执行request_B。

当请求 A 正在等待时，请求 B 可能会执行

这种情况不会发生，这基本上await意味着：坚持下去，直到请求 A 返回，然后再继续。换句话说，await对执行顺序没有影响。它只是将控制权交给其他人，以便其他人可以使用隐藏时间（在您的情况下，来自另一个（A，B）请求对的任何代码）。这就是为什么连续的代码行不一定连续执行的原因，将控制权交给其他协程（我们刚才提到的其他人），使用await允许该协程在 A 和 B 之间执行代码。

即使这有点不准确，您也可以记住：唯一将并行执行的代码是您自己安排的代码（在本例中使用asyncio.gather，安排多个(A, B)对并行执行）。

我知道我可以批量运行所有请求 A 调用，然后批量运行所有请求 B 调用，但由于我的用例特定的原因，我需要运行一批所有...

在这种特殊情况下，即使您可以运行一批A，然后运行一批B，我认为您的解决方案会更好，因为它以更简单的方式突出显示A和B之间的关系。

下面是一个代码示例，您可以运行它来尝试一下（它的作用与您在此处使用公共数学 API 所做的相同），它只需分两步计算“x*2+2”，第一个“*2” （相当于请求A），然后“+2”（相当于请求B）：

MATH_API_URL = "http://api.mathjs.org/v4"

from aiohttp import ClientSession
import asyncio

async def maths(session, url, expression):
    params = {"expr" : expression}
    print(f"\t> computing {expression}")
    async with session.get(url, params=params) as response:
        result = await response.text()
        print(f"\t< {expression} = {result}")
        return result

async def twice(session, x):
    return await maths(session, MATH_API_URL, f"2 * {x}")

async def plus_two(session, x):
    return await maths(session, MATH_API_URL, f"2 + {x}")

async def twice_plus_two(session, x):
    twice_x = await twice(session, x)
    return await plus_two(session, twice_x)

async def main(inputs):
    async with ClientSession() as session:
        return await asyncio.gather(*(twice_plus_two(session, x) for x in inputs))

inputs = list(range(3))
print([x*2+2 for x in inputs])
print(asyncio.run(main(inputs)))

此代码输出请求的安排顺序：

[2, 4, 6]    
    > computing 2 * 0    
    > computing 2 * 1    
    > computing 2 * 2    
    < 2 * 1 = 2    
    > computing 2 + 2
    < 2 * 0 = 0    
    > computing 2 + 0
    < 2 * 2 = 4    
    > computing 2 + 4
    < 2 + 2 = 4    
    < 2 + 4 = 6
    < 2 + 0 = 2
['2', '4', '6']

查看“*2”返回后如何安排“+2”。