### 使用Python 3实现并发请求并一次性发送5个请求 在Python中，可以通过多种方式来实现并发请求。以下是基于`concurrent.futures.ThreadPoolExecutor`模块的一个简单示例，该方法适用于中小型规模的并发请求场景。 #### 实现代码 以下是一个完整的代码示例，展示了如何通过线程池执行器（ThreadPoolExecutor）发出5个并发请求： ```python import requests from concurrent.futures import ThreadPoolExecutor, as_completed import time def fetch(url, data=None): """发送单个HTTP POST请求，并记录响应时间和状态""" start_time = time.time() try: response = requests.post(url, json=data) elapsed_time = time.time() - start_time return { "status_code": response.status_code, "elapsed_time": elapsed_time, "response_text": response.text[:100], # 截断显示前100字符 } except Exception as e: return {"error": str(e), "elapsed_time": time.time() - start_time} def main(urls, data_list): concurrency = min(len(urls), 5) # 设置最大并发数为5或URL列表长度 results = [] with ThreadPoolExecutor(max_workers=concurrency) as executor: futures = [ executor.submit(fetch, url, data) for url, data in zip(urls, data_list) ] for future in as_completed(futures): result = future.result() results.append(result) total_elapsed_time = sum(r["elapsed_time"] for r in results if "elapsed_time" in r) print(f"Total Requests: {len(results)}, Total Time Elapsed: {total_elapsed_time:.2f}s") return results if __name__ == "__main__": urls = ["https://httpbin.org/post"] * 5 # 测试用的POST URL data_list = [{"key": f"value_{i}"} for i in range(5)] # 请求数据 responses = main(urls, data_list) for idx, resp in enumerate(responses, start=1): print(f"Request {idx}: {resp}") ``` #### 说明 上述代码实现了以下几个功能： - 定义了一个`fetch`函数用于发送单一HTTP POST请求，并捕获可能发生的异常[^2]。 - `main`函数利用`ThreadPoolExecutor`创建了一个固定大小的工作线程池，其最大工作线程数被设定为5（即并发请求数），并通过`as_completed`迭代器实时跟踪已完成的任务结果[^2]。 - 打印每个请求的结果及其耗时，同时计算所有请求完成所需的总时间[^2]。 此方案适合于中小规模的并发需求，能够有效减少等待时间，提升效率[^1]。 #### 注意事项 尽管这种方法简单高效，但在实际部署到生产环境中时需注意以下几点： - **服务器负载**：过高的并发可能导致目标服务器资源紧张甚至崩溃，因此建议逐步增加并发数量以评估服务器性能[^2]。 - **错误处理与重试机制**：为了增强稳定性，可加入更完善的错误捕捉逻辑以及自动重试策略[^4]。 - **安全性考量**：如果涉及敏感操作，则务必遵循最佳实践保护传输中的数据[^3]。 --- ###