Apache HttpClient 4.5连接池：限制等待线程数及超额请求处理

原生配置是否支持？ #

Apache HttpClient 4.5的PoolingHttpClientConnectionManager并没有直接提供限制等待线程数量的配置项。它的setConnectionRequestTimeout(int)方法仅用于设置单个线程等待可用连接的超时时间，但所有请求线程都会进入等待队列，不会直接拒绝超出指定数量的等待请求。

装饰器模式结合信号量实现方案 #

要实现“连接池耗尽时允许最多n个线程等待，超出的请求立即失败”的需求，可通过装饰器模式包装原生HttpClientConnectionManager，结合信号量控制等待线程的数量。

实现思路 #

1. 创建一个装饰类，代理原有的HttpClientConnectionManager实例
2. 初始化一个Semaphore，许可数设为允许等待的最大线程数n
3. 在请求连接前尝试获取信号量许可：
   • 成功获取则继续执行连接请求，连接释放或请求取消时释放许可
   • 获取失败（超时或无可用许可）则直接抛出异常，拒绝当前请求

完整代码示例 #

import org.apache.http.conn.HttpClientConnectionManager;
import org.apache.http.conn.routing.HttpRoute;
import org.apache.http.conn.ConnectionRequest;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.io.IOException;

public class LimitedWaitConnectionManager implements HttpClientConnectionManager {
    private final HttpClientConnectionManager delegate;
    private final Semaphore waitSemaphore;
    private final long acquireTimeout;
    private final TimeUnit timeUnit;

    public LimitedWaitConnectionManager(HttpClientConnectionManager delegate, int maxWaitThreads, long acquireTimeout, TimeUnit timeUnit) {
        this.delegate = delegate;
        this.waitSemaphore = new Semaphore(maxWaitThreads);
        this.acquireTimeout = acquireTimeout;
        this.timeUnit = timeUnit;
    }

    @Override
    public ConnectionRequest requestConnection(HttpRoute route, Object state) {
        try {
            // 尝试获取信号量，超时则直接拒绝请求
            if (!waitSemaphore.tryAcquire(acquireTimeout, timeUnit)) {
                throw new IOException("Exceeded maximum waiting threads, request rejected");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IOException("Connection request interrupted", e);
        }

        // 包装原ConnectionRequest，确保信号量在所有场景下都能释放
        ConnectionRequest originalRequest = delegate.requestConnection(route, state);
        return new ConnectionRequest() {
            @Override
            public boolean cancel() {
                boolean canceled = originalRequest.cancel();
                if (canceled) {
                    waitSemaphore.release();
                }
                return canceled;
            }

            @Override
            public void awaitConnection() throws InterruptedException {
                try {
                    originalRequest.awaitConnection();
                } catch (InterruptedException | RuntimeException e) {
                    waitSemaphore.release();
                    throw e;
                }
            }
        };
    }

    // 以下方法直接委托给原连接池，确保原有功能不受影响
    @Override
    public void releaseConnection(org.apache.http.HttpClientConnection conn, Object state, long keepalive, TimeUnit tunit) {
        try {
            delegate.releaseConnection(conn, state, keepalive, tunit);
        } finally {
            waitSemaphore.release();
        }
    }

    @Override
    public void connect(org.apache.http.HttpClientConnection conn, HttpRoute route, int connectTimeout, org.apache.http.protocol.HttpContext context) throws IOException {
        delegate.connect(conn, route, connectTimeout, context);
    }

    @Override
    public void upgrade(org.apache.http.HttpClientConnection conn, HttpRoute route, org.apache.http.protocol.HttpContext context) throws IOException {
        delegate.upgrade(conn, route, context);
    }

    @Override
    public void routeComplete(org.apache.http.HttpClientConnection conn, HttpRoute route, org.apache.http.protocol.HttpContext context) throws IOException {
        delegate.routeComplete(conn, route, context);
    }

    @Override
    public void shutdown() {
        delegate.shutdown();
    }

    @Override
    public void closeExpiredConnections() {
        delegate.closeExpiredConnections();
    }

    @Override
    public void closeIdleConnections(long idletime, TimeUnit tunit) {
        delegate.closeIdleConnections(idletime, tunit);
    }
}

使用示例 #

// 初始化原生连接池，配置基础连接数
PoolingHttpClientConnectionManager originalManager = new PoolingHttpClientConnectionManager();
originalManager.setMaxTotal(20); // 全局最大连接数
originalManager.setDefaultMaxPerRoute(10); // 单路由最大连接数

// 包装为带等待限制的连接池：允许最多5个线程等待，超时时间3秒
HttpClientConnectionManager limitedManager = new LimitedWaitConnectionManager(
        originalManager,
        5,
        3000,
        TimeUnit.MILLISECONDS
);

// 构建最终的HttpClient实例
CloseableHttpClient httpClient = HttpClients.custom()
        .setConnectionManager(limitedManager)
        .build();

关键注意事项 #

• 信号量的许可数maxWaitThreads需根据系统负载和业务并发量合理设置，避免过小导致请求被过度拒绝
• 必须确保信号量在所有异常、取消场景下都能正确释放，防止许可泄漏导致后续请求无法获取资源
• 原生连接池的setConnectionRequestTimeout无需额外设置，装饰类已通过信号量的超时逻辑实现了等待控制

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