python - 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

Per each task consisting of a pair of requests, is Request A guaranteed to return before Request B is sent?

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

while Request A is being awaited, Request B may execute

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

即使这有点不准确，您也可以记住:唯一将并行执行的代码是您自己安排的代码(在这种情况下使用 asyncio.gather ，安排多个 (A, B) 对并行执行) .

I understand that I can run all Request A calls in a batch, then run all Request B calls in a batch, but for reasons specific to my use case, I need to run a batch of all ...

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

这是您可以运行来尝试的代码示例(它与您在此处使用公共(public)数学 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”。