如何使用async/await加速请求处理同时保留原始数据ID顺序？

问题 #

我想用async模拟向网站发送请求并等待返回结果，要求按原始ID顺序输出结果而非异步执行后的乱序。当前代码如下：

import asyncio
import random

class Count: #Assume data only hold the data ID number
  count = 0

def output(count):
  print(count) #Assume storing the ID and content of the website into a database

async def count():
  while True:
    Count.count = Count.count + 1 #Update the data reference ID by increasing one
    count = Count.count #Assume ID is the variable that will be pass to a website
    await asyncio.sleep(random.randint(1, 10)) #Simluate connect to the website and passing the ID variables and return the result from the website
    output(count)
    if Count.count>9:
      break

async def main():
  await asyncio.gather(count(), count(),count(),count())
if __name__ == "__main__":
  asyncio.run(main())

执行结果为：

3
1
5
4
2
6
8
7
10
9

该结果不符合预期，虽然async/await提升了处理速度，但牺牲了数据顺序。我希望在保留异步加速的同时，维持ID从1到10的输出顺序。我可以先保存所有数据再按ID排序，但有没有更优的方式，比如让第二个异步count()协程的output()调用在第一个的output()完成后再执行？预期输出为：

1
2
3
4
5
6
7
8
9
10

解决方案 #

要实现异步请求+按ID顺序输出，无需等待所有请求完成再排序，也不用让协程输出互相阻塞，可通过拆分"请求任务"与"输出逻辑"的方式实现，兼顾性能与顺序要求。

方案一：基于gather的有序结果返回 #

利用asyncio.gather的特性——它会严格按照传入任务列表的顺序返回结果，即使部分请求先完成，最终结果列表仍保持原任务顺序。

import asyncio
import random

def output(count):
    print(count)  # 模拟将ID和内容存入数据库

async def fetch_data(id_num):
    # 模拟网站请求的随机耗时
    await asyncio.sleep(random.randint(1, 10))
    return id_num

async def main():
    # 生成1到10的请求任务列表
    tasks = [fetch_data(i) for i in range(1, 11)]
    # 异步执行所有任务，按原顺序获取结果
    results = await asyncio.gather(*tasks)
    # 按ID顺序输出结果
    for res in results:
        output(res)

if __name__ == "__main__":
    asyncio.run(main())

方案二：基于队列的实时有序输出 #

如果不想等待所有请求完成再输出，而是希望前面的ID请求完成后立刻输出，可借助异步队列实现：单独启动一个协程按顺序消费队列，所有请求完成后将结果放入队列，确保输出顺序。

import asyncio
import random

async def ordered_output(queue):
    # 按ID顺序从队列取结果输出
    for _ in range(1, 11):
        res = await queue.get()
        print(res)
        queue.task_done()

async def fetch_data(id_num, queue):
    # 模拟网站请求耗时
    await asyncio.sleep(random.randint(1, 10))
    # 请求完成后将结果放入队列
    await queue.put(id_num)

async def main():
    queue = asyncio.Queue()
    # 启动有序输出协程
    output_task = asyncio.create_task(ordered_output(queue))
    # 启动所有请求协程
    fetch_tasks = [fetch_data(i, queue) for i in range(1, 11)]
    await asyncio.gather(*fetch_tasks)
    await queue.join()
    output_task.cancel()

if __name__ == "__main__":
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者jguy