如何在Locust所用的urllib3中实现类似PooledConnectionLifetime的连接生命周期？

问题原因分析 #

你之前自定义PoolManager子类未被调用，核心原因是Locust的HttpUser基于requests库封装，requests默认使用自身的HTTPAdapter和urllib3默认PoolManager，直接替换子类不会被requests加载，必须通过自定义并挂载HTTPAdapter才能生效。

另外关于start_connect()的执行时机：它仅在首次创建连接时触发，从连接池取出连接时不会重复执行，因此conn.timeout.get_connect_duration()可以准确获取连接从创建到当前的总时长，适合用来判断是否超过生命周期。

解决方案 #

方案1：利用urllib3内置max_age参数（推荐） #

urllib3的ConnectionPool原生支持max_age参数，设置后连接在池中存活超过该时长时，会被自动丢弃并重建。通过自定义HTTPAdapter让requests使用该配置：

from locust import HttpUser, task
from requests.adapters import HTTPAdapter
from urllib3 import PoolManager

class LifespanHTTPAdapter(HTTPAdapter):
    def __init__(self, max_lifetime, *args, **kwargs):
        self.max_lifetime = max_lifetime
        super().__init__(*args, **kwargs)

    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
        self.poolmanager = PoolManager(
            num_pools=connections,
            maxsize=maxsize,
            block=block,
            # 配置连接最大存活时长（单位：秒）
            connection_pool_kwargs={"max_age": self.max_lifetime},
            **pool_kwargs
        )

class LoadTestUser(HttpUser):
    def on_start(self):
        # 设置连接生命周期为30秒
        adapter = LifespanHTTPAdapter(max_lifetime=30)
        # 为HTTP和HTTPS请求挂载适配器
        self.client.mount("http://", adapter)
        self.client.mount("https://", adapter)

    @task
    def sample_request(self):
        self.client.get("/your-api-endpoint")

方案2：自定义连接池实现更灵活的生命周期控制 #

如果需要定制化逻辑（比如关闭前的额外操作），可以重写urllib3的ConnectionPool类，手动检查连接存活时长：

from locust import HttpUser, task
from requests.adapters import HTTPAdapter
from urllib3 import PoolManager, HTTPConnectionPool, HTTPSConnectionPool
import time

class LifespanHTTPConnectionPool(HTTPConnectionPool):
    def __init__(self, *args, max_lifetime=30, **kwargs):
        self.max_lifetime = max_lifetime
        super().__init__(*args, **kwargs)

    def _get_conn(self, timeout=None):
        conn = super()._get_conn(timeout=timeout)
        # 检查连接是否超过生命周期
        if time.time() - conn.timeout.connect_start > self.max_lifetime:
            conn.close()
            # 重新获取新连接
            conn = super()._get_conn(timeout=timeout)
        return conn

class LifespanHTTPSConnectionPool(HTTPSConnectionPool):
    def __init__(self, *args, max_lifetime=30, **kwargs):
        self.max_lifetime = max_lifetime
        super().__init__(*args, **kwargs)

    def _get_conn(self, timeout=None):
        conn = super()._get_conn(timeout=timeout)
        if time.time() - conn.timeout.connect_start > self.max_lifetime:
            conn.close()
            conn = super()._get_conn(timeout=timeout)
        return conn

class CustomLifespanAdapter(HTTPAdapter):
    def __init__(self, max_lifetime, *args, **kwargs):
        self.max_lifetime = max_lifetime
        super().__init__(*args, **kwargs)

    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
        self.poolmanager = PoolManager(
            num_pools=connections,
            maxsize=maxsize,
            block=block,
            connection_pool_kwargs={"max_lifetime": self.max_lifetime},
            **pool_kwargs
        )
        # 替换为自定义的连接池类
        self.poolmanager.pool_classes_by_scheme = {
            "http": LifespanHTTPConnectionPool,
            "https": LifespanHTTPSConnectionPool,
        }

class LoadTestUser(HttpUser):
    def on_start(self):
        adapter = CustomLifespanAdapter(max_lifetime=30)
        self.client.mount("http://", adapter)
        self.client.mount("https://", adapter)

    @task
    def sample_request(self):
        self.client.get("/your-api-endpoint")

验证方法 #

可以在自定义连接池的_get_conn方法中添加日志（比如print(f"Reusing connection created at {conn.timeout.connect_start}")），观察连接是否在超时后被重建，确认功能生效。

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