Python多线程场景下requests timeout不生效问题解决方案

requests.get自带的timeout参数并非请求全生命周期的硬超时，仅覆盖两个阶段的空闲等待阈值：

• TCP连接建立阶段的最长等待时间
• 连接建立后，两次相邻socket数据包到达的最大间隔时间

以下常见场景会导致该参数完全失效，请求无限挂起：

• 系统底层DNS解析阻塞：requests默认调用操作系统阻塞式DNS查询接口，该流程不受timeout参数控制
• 服务端/Cloudflare拦截请求后，以小于timeout值的间隔持续发送无意义TCP保活报文、零散响应字节，每次收到数据都会重置读取超时计时器，永远无法触发超时逻辑
• 部分旧版本依赖库对SSL握手阶段的超时覆盖存在缺陷，握手异常时会直接阻塞

你代码中异常块永远不触发、执行流卡在requests.get行，就是因为上述场景下requests没有抛出任何异常，一直处于等待状态。

方案1：基于线程池的任务级硬超时（通用无坑，推荐优先使用） #

用标准库concurrent.futures将请求提交到独立线程池执行，给整个请求任务设置固定超时阈值，超时后直接放弃等待结果，工作线程立刻进入下一轮消费逻辑，不会被卡住。
改造后的可运行代码如下：

import queue
import threading
import requests
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError

json_queue = queue.Queue()
# 独立请求线程池，和消费逻辑隔离
request_executor = ThreadPoolExecutor(max_workers=4)
# 全请求硬超时，单位：秒
TOTAL_HARD_TIMEOUT = 3

def send_request(url: str):
    # 保留requests自带超时作为第一层防护，连接超时2s，读取超时3s
    return requests.get(url=url, timeout=(2, 3))

def event_collect_thread():
    while True:
        for event in some_external_event:
            target_url = event
            json_queue.put(target_url)

def request_consumer_thread():
    while True:
        current_url = json_queue.get()
        try:
            task = request_executor.submit(send_request, current_url)
            resp = task.result(timeout=TOTAL_HARD_TIMEOUT)
            # 此处编写请求成功后的处理逻辑
            # print(resp.status_code, resp.text)
        except FuturesTimeoutError:
            task.cancel()
            # 超时后可在此处编写重试、日志记录、直接跳过逻辑
            print(f"请求{current_url}超时，已强制终止")
        except Exception as e:
            # 捕获连接错误、HTTP错误等其他异常
            print(f"请求{current_url}出错：{str(e)}")
        finally:
            json_queue.task_done()

if __name__ == "__main__":
    t1 = threading.Thread(target=event_collect_thread, daemon=True)
    t2 = threading.Thread(target=request_consumer_thread, daemon=True)
    t1.start()
    t2.start()
    # 阻塞主线程防止退出
    t1.join()

说明：Python没有安全强制终止运行中线程的机制，线程池中卡住的请求任务不会被立刻杀掉，但它不会阻塞消费主流程，卡住的连接会被操作系统在TCP保活周期到期后自动回收，资源占用可控。

方案2：自定义HTTP适配器，从socket层强制超时 #

如果不想引入额外线程，可以自定义requests的HTTP适配器，直接给底层socket设置全局收发超时，从系统调用层面强制所有网络操作超时，覆盖DNS、SSL握手、数据收发全流程：

import requests
import socket
import struct
from requests.adapters import HTTPAdapter

class SocketHardTimeoutAdapter(HTTPAdapter):
    def __init__(self, *args, **kwargs):
        self.hard_timeout = kwargs.pop("total_timeout", 3)
        super().__init__(*args, **kwargs)

    def init_poolmanager(self, *args, **kwargs):
        # 将秒数转换为socket要求的timeval结构体
        timeval_pack = struct.pack("ll", self.hard_timeout, 0)
        kwargs["socket_options"] = [
            (socket.SOL_SOCKET, socket.SO_RCVTIMEO, timeval_pack),
            (socket.SOL_SOCKET, socket.SO_SNDTIMEO, timeval_pack),
            (socket.SOL_TCP, socket.TCP_USER_TIMEOUT, self.hard_timeout * 1000)
        ]
        super().init_poolmanager(*args, **kwargs)

# 初始化全局复用的session
http_session = requests.Session()
timeout_adapter = SocketHardTimeoutAdapter(total_timeout=3)
http_session.mount("http://", timeout_adapter)
http_session.mount("https://", timeout_adapter)

# 后续直接用http_session.get发起请求即可，超时会强制抛出异常
# resp = http_session.get(url, timeout=3)

• 不要直接使用裸requests.get，每次调用都会新建TCP连接，开销大且无法统一配置规则，优先使用requests.Session()复用连接
• 重试逻辑需要设置最大重试次数，避免异常URL无限重试占用资源
• 工作线程建议设置为daemon=True，避免主进程退出时被残留阻塞线程卡住
• 针对Cloudflare拦截场景，可以补充正常浏览器的User-Agent、TLS指纹等信息，从源头减少被拦截挂起的概率

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