python第三方库可以处理多线程请求接口结果顺序问题？

pip install grequests

import grequests
import time

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
TOTAL_REQUESTS = 20
TIMEOUT = 5

def request_api(index: int) -> tuple:
    """单个请求函数：返回（索引，是否成功，结果信息）"""
    url = API_URL.format(index % 10 + 1)
    try:
        # 用 grequests.get 替代 requests.get，支持协程并发
        response = grequests.get(url, timeout=TIMEOUT)
        response.send()  # 发送请求（非阻塞）
        response = response.response  # 获取响应对象
        response.raise_for_status()
        return (index, True, f"响应：{response.json()['title'][:20]}...")
    except Exception as e:
        return (index, False, f"失败：{str(e)[:30]}")

if __name__ == "__main__":
    start_time = time.time()

    # 1. 构造请求列表（按顺序提交）
    requests_list = [grequests.get(API_URL.format(i % 10 + 1), timeout=TIMEOUT) for i in range(TOTAL_REQUESTS)]
    
    # 2. 并发执行，map 方法保证结果顺序与请求顺序一致
    responses = grequests.map(requests_list, size=10)  # size=并发数（类似线程池大小）

    # 3. 按顺序处理结果（responses 顺序 = 请求提交顺序）
    print("grequests 协程并发 - 按请求顺序输出：")
    for idx, response in enumerate(responses):
        if response and response.status_code == 200:
            msg = f"响应：{response.json()['title'][:20]}..."
            print(f"任务[{idx}]：✅ {msg}")
        else:
            err_msg = response.reason if response else "请求超时/失败"
            print(f"任务[{idx}]：❌ 失败：{err_msg[:30]}")

    total_cost = round(time.time() - start_time, 3)
    print(f"\n总耗时：{total_cost}s")

pip install httpx

import httpx
import asyncio
import time

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
TOTAL_REQUESTS = 20
TIMEOUT = 5
CONCURRENT_NUM = 10  # 并发数

async def request_api(client: httpx.AsyncClient, index: int) -> tuple:
    """异步请求函数"""
    url = API_URL.format(index % 10 + 1)
    try:
        response = await client.get(url, timeout=TIMEOUT)
        response.raise_for_status()
        data = response.json()
        return (index, True, f"响应：{data['title'][:20]}...")
    except Exception as e:
        return (index, False, f"失败：{str(e)[:30]}")

async def main():
    start_time = time.time()

    # 1. 创建异步客户端（支持连接池复用）
    async with httpx.AsyncClient(limits=httpx.Limits(max_connections=CONCURRENT_NUM)) as client:
        # 2. 构造异步任务列表（按顺序提交）
        tasks = [request_api(client, i) for i in range(TOTAL_REQUESTS)]
        # 3. 并发执行，gather 保证结果顺序与任务顺序一致
        results = await asyncio.gather(*tasks)

    # 4. 按顺序输出结果
    print("httpx 异步并发 - 按请求顺序输出：")
    for idx, is_success, msg in results:
        print(f"任务[{idx}]：{'✅' if is_success else '❌'} {msg}")

    total_cost = round(time.time() - start_time, 3)
    print(f"\n总耗时：{total_cost}s")

if __name__ == "__main__":
    # 兼容 Windows 系统
    import sys
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())

pip install tenacity

import requests
from concurrent.futures import ThreadPoolExecutor
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import time

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
THREAD_NUM = 10
TOTAL_REQUESTS = 20
TIMEOUT = 5

# 配置重试策略：失败最多重试 2 次，每次间隔 1s、2s（指数退避）
@retry(
    stop=stop_after_attempt(2),  # 最多重试 2 次
    wait=wait_exponential(multiplier=1, min=1, max=5),  # 重试间隔：1s、2s、4s...
    retry=retry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError))  # 仅对超时/连接错误重试
)
def request_api(index: int) -> tuple:
    """单个请求函数（带重试）"""
    url = API_URL.format(index % 10 + 1)
    try:
        response = requests.get(url, timeout=TIMEOUT)
        response.raise_for_status()
        return (index, True, f"响应：{response.json()['title'][:20]}...")
    except Exception as e:
        return (index, False, f"失败：{str(e)[:30]}")

if __name__ == "__main__":
    start_time = time.time()

    # 结合 ThreadPoolExecutor 有序收集结果
    with ThreadPoolExecutor(max_workers=THREAD_NUM) as executor:
        future_list = [executor.submit(request_api, i) for i in range(TOTAL_REQUESTS)]
        print("ThreadPoolExecutor + tenacity 重试 - 按请求顺序输出：")
        for future in future_list:  # 按提交顺序获取结果（有序）
            idx, is_success, msg = future.result()
            print(f"任务[{idx}]：{'✅' if is_success else '❌'} {msg}")

    total_cost = round(time.time() - start_time, 3)
    print(f"\n总耗时：{total_cost}s")

pip install trio httpx-trio  # httpx-trio 是 httpx 的 trio 适配库

import trio
import httpx
import time

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
TOTAL_REQUESTS = 20
TIMEOUT = 5
CONCURRENT_NUM = 10  # 最大并发数

async def request_api(client: httpx.AsyncClient, index: int) -> tuple:
    """异步请求函数"""
    url = API_URL.format(index % 10 + 1)
    try:
        response = await client.get(url, timeout=TIMEOUT)
        response.raise_for_status()
        data = response.json()
        return (index, True, f"响应：{data['title'][:20]}...")
    except Exception as e:
        return (index, False, f"失败：{str(e)[:30]}")

async def main():
    start_time = time.time()

    # 1. 创建异步客户端（结合 trio）
    async with httpx.AsyncClient(limits=httpx.Limits(max_connections=CONCURRENT_NUM)) as client:
        # 2. 构造任务列表（按顺序），用 trio.gather 并发执行并保证顺序
        results = await trio.gather(*[request_api(client, i) for i in range(TOTAL_REQUESTS)])

    # 3. 按顺序输出结果
    print("trio + httpx 异步并发 - 按请求顺序输出：")
    for idx, is_success, msg in results:
        print(f"任务[{idx}]：{'✅' if is_success else '❌'} {msg}")

    total_cost = round(time.time() - start_time, 3)
    print(f"\n总耗时：{total_cost}s")

if __name__ == "__main__":
    trio.run(main())  # trio 自带事件循环，无需手动启动