测试场景下如何发送大量HTTP请求且无需等待响应？

嘿，这个问题我太熟了——同步阻塞的requests循环处理上万请求肯定慢得让人抓狂，毕竟每个请求都要等服务器响应完才会继续下一个。既然你完全不关心响应，那我们可以直接跳过等待响应的步骤，甚至不用建立完整的HTTP会话，直接把请求发出去就行。下面给你几个从快到慢（但复杂度也从高到低）的方案：

方案1：直接用Socket发送原始HTTP请求（最快，最底层） #

因为我们不需要响应，所以只要把GET请求的字节流发出去，立刻关闭连接就行，不用等服务器回包。这种方式跳过了所有HTTP库的响应处理逻辑，效率拉满。

import socket
from concurrent.futures import ThreadPoolExecutor

def send_raw_request(url, param):
    # 手动解析URL，拆分主机和路径
    url_parts = url.split('://')[1].split('/')
    host = url_parts[0]
    path = '/' + '/'.join(url_parts[1:]) if len(url_parts) > 1 else '/'
    
    # 构建GET请求的参数字符串
    param_str = '&'.join([f"{k}={v}" for k, v in param.items()])
    # 组装完整的HTTP请求头
    request = (
        f"GET {path}?{param_str} HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        "Connection: close\r\n\r\n"
    ).encode('utf-8')
    
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(0.5)  # 设置短超时，避免无意义的等待
        sock.connect((host, 80))  # HTTPS的话改端口为443，还要加SSL包装
        sock.sendall(request)
        sock.close()
    except Exception:
        # 不关心错误，直接忽略
        pass

def fire(urls, params):
    # 用线程池并发发送，IO密集型场景线程池能大幅提升效率
    with ThreadPoolExecutor(max_workers=50) as executor:
        for url in urls:
            for param in params:
                executor.submit(send_raw_request, url, param)

注：如果是HTTPS请求，需要用ssl.wrap_socket()对socket进行包装，会增加一点开销，但还是远快于requests。

方案2：用aiohttp异步发送，跳过响应读取 #

aiohttp是异步HTTP库，比同步requests高效得多，我们可以创建请求任务后立刻取消，完全跳过响应等待环节。

import asyncio
import aiohttp

async def send_async_request(session, url, param):
    try:
        # 创建请求任务，但不等响应就取消
        task = asyncio.create_task(session.get(url, params=param))
        task.cancel()
        await task
    except asyncio.CancelledError:
        # 捕获取消异常，直接忽略
        pass
    except Exception:
        pass

async def fire_async(urls, params):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            for param in params:
                tasks.append(send_async_request(session, url, param))
        # 并发执行所有任务，忽略异常
        await asyncio.gather(*tasks, return_exceptions=True)

def fire(urls, params):
    asyncio.run(fire_async(urls, params))

说明：异步模型在处理大量IO请求时，资源利用率比线程池更高，适合超大规模的请求场景。

方案3：用requests+线程池（最简单，改动最小） #

如果你不想换库，只想在原有代码基础上提速，用线程池并发发送是最省心的方式，多个请求可以同时进行，不用逐个等待。

import requests
from concurrent.futures import ThreadPoolExecutor

def send_request(url, param):
    try:
        # 设置极短超时，开启stream模式避免读取响应内容
        requests.get(url, params=param, timeout=0.1, stream=True)
        # 直接结束，不处理响应
        return
    except Exception:
        pass

def fire(urls, params):
    with ThreadPoolExecutor(max_workers=100) as executor:
        for url in urls:
            for param in params:
                executor.submit(send_request, url, param)

说明：这个方案几乎是在你原有代码上做最小改动，线程池的并发能让速度比同步循环提升数倍。

总结 #

• 追求极致速度选方案1，完全跳过HTTP库的冗余处理；
• 想拥抱异步生态选方案2，资源利用率最高；
• 想最小改动原有代码选方案3，省心省力。

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