基于Tornado实现多异步HTTP请求及批量响应处理的技术问询

嘿，我来帮你搞定这个需求！Tornado本身就是为异步场景设计的，要实现“发起多个异步HTTP请求，等所有响应返回后再处理后续业务”非常合适。下面我给你一套完整的解决方案，包含代码示例和关键细节说明。

核心思路 #

1. 用Tornado的AsyncHTTPClient发起单个异步HTTP请求
2. 借助tornado.gen.multi（或者Python标准库的asyncio.gather）来并发执行所有请求，等待全部完成
3. 拿到所有响应结果后，统一进行后续业务处理

完整代码示例 #

假设你的url_handlers是包含请求配置的列表（支持GET/POST、自定义头、请求体等），代码如下：

import tornado.ioloop
import tornado.httpclient
from tornado.gen import coroutine
import json

# 你的url_handler列表，每个元素包含请求的配置信息
url_handlers = [
    {"url": "https://api.example.com/data1", "method": "GET", "headers": {"Accept": "application/json"}},
    {"url": "https://api.example.com/data2", "method": "POST", "body": json.dumps({"key": "test"}).encode("utf-8"), "headers": {"Content-Type": "application/json"}},
    {"url": "https://api.example.com/data3"},  # 默认GET请求
]

async def fetch_single_request(handler):
    # 初始化异步HTTP客户端
    http_client = tornado.httpclient.AsyncHTTPClient()
    
    try:
        # 构建HTTP请求对象，支持自定义方法、头、请求体、超时等
        request = tornado.httpclient.HTTPRequest(
            url=handler["url"],
            method=handler.get("method", "GET"),
            body=handler.get("body"),
            headers=handler.get("headers", {}),
            connect_timeout=5,  # 连接超时5秒
            request_timeout=10  # 请求超时10秒
        )
        
        # 发起异步请求并等待响应
        response = await http_client.fetch(request)
        
        # 对单个响应做预处理（比如解析JSON）
        try:
            response_data = json.loads(response.body)
        except json.JSONDecodeError:
            response_data = response.body.decode("utf-8")
        
        return {
            "url": handler["url"],
            "status": response.code,
            "data": response_data,
            "success": True
        }
    
    except tornado.httpclient.HTTPError as e:
        # 处理HTTP错误（比如404、500等状态码）
        return {
            "url": handler["url"],
            "status": e.code,
            "error": str(e),
            "success": False
        }
    except Exception as e:
        # 处理其他异常（比如网络连接失败）
        return {
            "url": handler["url"],
            "error": str(e),
            "success": False
        }

async def main():
    print("开始批量发起异步请求...")
    
    # 并发执行所有请求，等待全部完成后拿到结果列表
    all_results = await tornado.gen.multi([fetch_single_request(handler) for handler in url_handlers])
    
    # --------------------------
    # 所有响应返回后，这里写你的后续业务逻辑
    # --------------------------
    print("\n所有请求完成，开始处理后续业务：")
    
    # 统计成功和失败的请求
    success_count = sum(1 for res in all_results if res["success"])
    fail_count = len(all_results) - success_count
    print(f"请求完成统计：成功{success_count}个，失败{fail_count}个")
    
    # 处理成功的响应数据
    successful_data = [res["data"] for res in all_results if res["success"]]
    if successful_data:
        print("\n成功获取的数据：")
        for idx, data in enumerate(successful_data):
            print(f"第{idx+1}条数据：{data}")
    
    # 处理失败的请求（比如记录日志、重试等）
    failed_requests = [res for res in all_results if not res["success"]]
    if failed_requests:
        print("\n失败的请求：")
        for res in failed_requests:
            print(f"URL: {res['url']}，原因: {res.get('error', '未知错误')}")

if __name__ == "__main__":
    # 启动Tornado的IO循环，运行main函数直到完成
    tornado.ioloop.IOLoop.current().run_sync(main)

关键细节说明 #

• 并发控制：tornado.gen.multi会同时发起所有请求，如果你有上百个请求，建议用tornado.queues.Queue来限制并发数（比如每次只发起10个请求），避免触发目标服务器的限流或者耗尽本地资源。
• 错误隔离：单个请求失败不会影响其他请求，每个请求的异常都被单独捕获，返回包含错误信息的结果。
• 响应预处理：在fetch_single_request里可以提前对每个响应做解析（比如JSON转字典），后续业务逻辑直接用处理好的数据即可。
• 超时设置：一定要给请求加上超时参数，防止某个请求长时间挂起导致整个任务卡住。

如果你的url_handlers只是简单的URL字符串列表（不需要自定义请求方法等），可以把fetch_single_request简化成只接收URL参数的版本，代码会更简洁。

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