聊聊httpclient的CPool

序

本文主要研究一下httpclient的CPool

ConnPool

org/apache/http/pool/ConnPool.java

public interface ConnPool<T, E> {/*** Attempts to lease a connection for the given route and with the given* state from the pool.** @param route route of the connection.* @param state arbitrary object that represents a particular state*  (usually a security principal or a unique token identifying*  the user whose credentials have been used while establishing the connection).*  May be {@code null}.* @param callback operation completion callback.** @return future for a leased pool entry.*/Future<E> lease(final T route, final Object state, final FutureCallback<E> callback);/*** Releases the pool entry back to the pool.** @param entry pool entry leased from the pool* @param reusable flag indicating whether or not the released connection*   is in a consistent state and is safe for further use.*/void release(E entry, boolean reusable);}

ConnPool定义了lease及release方法，其中定义了两个泛型，T表示route，E表示poolEntry

ConnPoolControl

public interface ConnPoolControl<T> {void setMaxTotal(int max);int getMaxTotal();void setDefaultMaxPerRoute(int max);int getDefaultMaxPerRoute();void setMaxPerRoute(final T route, int max);int getMaxPerRoute(final T route);PoolStats getTotalStats();PoolStats getStats(final T route);}

ConnPoolControl接口定义了设置和访问maxTotal、defaultMaxPerRoute及PoolStats的方法

AbstractConnPool

org/apache/http/pool/AbstractConnPool.java

@Contract(threading = ThreadingBehavior.SAFE_CONDITIONAL)
public abstract class AbstractConnPool<T, C, E extends PoolEntry<T, C>>implements ConnPool<T, E>, ConnPoolControl<T> {private final Lock lock;private final Condition condition;private final ConnFactory<T, C> connFactory;private final Map<T, RouteSpecificPool<T, C, E>> routeToPool;private final Set<E> leased;private final LinkedList<E> available;private final LinkedList<Future<E>> pending;private final Map<T, Integer> maxPerRoute;private volatile boolean isShutDown;private volatile int defaultMaxPerRoute;private volatile int maxTotal;private volatile int validateAfterInactivity;public AbstractConnPool(final ConnFactory<T, C> connFactory,final int defaultMaxPerRoute,final int maxTotal) {super();this.connFactory = Args.notNull(connFactory, "Connection factory");this.defaultMaxPerRoute = Args.positive(defaultMaxPerRoute, "Max per route value");this.maxTotal = Args.positive(maxTotal, "Max total value");this.lock = new ReentrantLock();this.condition = this.lock.newCondition();this.routeToPool = new HashMap<T, RouteSpecificPool<T, C, E>>();this.leased = new HashSet<E>();this.available = new LinkedList<E>();this.pending = new LinkedList<Future<E>>();this.maxPerRoute = new HashMap<T, Integer>();}/*** Creates a new entry for the given connection with the given route.*/protected abstract E createEntry(T route, C conn);//......}

AbstractConnPool声明实现ConnPool、ConnPoolControl接口，它定义E必须继承PoolEntry，同时定义了泛型C，表示connectionType

shutdown

    public void shutdown() throws IOException {if (this.isShutDown) {return ;}this.isShutDown = true;this.lock.lock();try {for (final E entry: this.available) {entry.close();}for (final E entry: this.leased) {entry.close();}for (final RouteSpecificPool<T, C, E> pool: this.routeToPool.values()) {pool.shutdown();}this.routeToPool.clear();this.leased.clear();this.available.clear();} finally {this.lock.unlock();}}

shutdown方法会遍历available、leased挨个执行close，然后遍历routeToPool挨个执行shutdown

lease方法

    public Future<E> lease(final T route, final Object state, final FutureCallback<E> callback) {Args.notNull(route, "Route");Asserts.check(!this.isShutDown, "Connection pool shut down");return new Future<E>() {private final AtomicBoolean cancelled = new AtomicBoolean(false);private final AtomicBoolean done = new AtomicBoolean(false);private final AtomicReference<E> entryRef = new AtomicReference<E>(null);@Overridepublic boolean cancel(final boolean mayInterruptIfRunning) {if (done.compareAndSet(false, true)) {cancelled.set(true);lock.lock();try {condition.signalAll();} finally {lock.unlock();}if (callback != null) {callback.cancelled();}return true;}return false;}@Overridepublic boolean isCancelled() {return cancelled.get();}@Overridepublic boolean isDone() {return done.get();}@Overridepublic E get() throws InterruptedException, ExecutionException {try {return get(0L, TimeUnit.MILLISECONDS);} catch (final TimeoutException ex) {throw new ExecutionException(ex);}}@Overridepublic E get(final long timeout, final TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {for (;;) {synchronized (this) {try {final E entry = entryRef.get();if (entry != null) {return entry;}if (done.get()) {throw new ExecutionException(operationAborted());}final E leasedEntry = getPoolEntryBlocking(route, state, timeout, timeUnit, this);if (validateAfterInactivity > 0)  {if (leasedEntry.getUpdated() + validateAfterInactivity <= System.currentTimeMillis()) {if (!validate(leasedEntry)) {leasedEntry.close();release(leasedEntry, false);continue;}}}if (done.compareAndSet(false, true)) {entryRef.set(leasedEntry);done.set(true);onLease(leasedEntry);if (callback != null) {callback.completed(leasedEntry);}return leasedEntry;} else {release(leasedEntry, true);throw new ExecutionException(operationAborted());}} catch (final IOException ex) {if (done.compareAndSet(false, true)) {if (callback != null) {callback.failed(ex);}}throw new ExecutionException(ex);}}}}};}

lease方法主要是get及cancel，其中get方法主要是执行getPoolEntryBlocking，对于validateAfterInactivity大于0的则根据判断是否需要validate，若需要且validate失败则执行leasedEntry.close()及release方法

getPoolEntryBlocking

org/apache/http/pool/AbstractConnPool.java

    private E getPoolEntryBlocking(final T route, final Object state,final long timeout, final TimeUnit timeUnit,final Future<E> future) throws IOException, InterruptedException, ExecutionException, TimeoutException {Date deadline = null;if (timeout > 0) {deadline = new Date (System.currentTimeMillis() + timeUnit.toMillis(timeout));}this.lock.lock();try {final RouteSpecificPool<T, C, E> pool = getPool(route);E entry;for (;;) {Asserts.check(!this.isShutDown, "Connection pool shut down");if (future.isCancelled()) {throw new ExecutionException(operationAborted());}for (;;) {entry = pool.getFree(state);if (entry == null) {break;}if (entry.isExpired(System.currentTimeMillis())) {entry.close();}if (entry.isClosed()) {this.available.remove(entry);pool.free(entry, false);} else {break;}}if (entry != null) {this.available.remove(entry);this.leased.add(entry);onReuse(entry);return entry;}// New connection is neededfinal int maxPerRoute = getMax(route);// Shrink the pool prior to allocating a new connectionfinal int excess = Math.max(0, pool.getAllocatedCount() + 1 - maxPerRoute);if (excess > 0) {for (int i = 0; i < excess; i++) {final E lastUsed = pool.getLastUsed();if (lastUsed == null) {break;}lastUsed.close();this.available.remove(lastUsed);pool.remove(lastUsed);}}if (pool.getAllocatedCount() < maxPerRoute) {final int totalUsed = this.leased.size();final int freeCapacity = Math.max(this.maxTotal - totalUsed, 0);if (freeCapacity > 0) {final int totalAvailable = this.available.size();if (totalAvailable > freeCapacity - 1) {if (!this.available.isEmpty()) {final E lastUsed = this.available.removeLast();lastUsed.close();final RouteSpecificPool<T, C, E> otherpool = getPool(lastUsed.getRoute());otherpool.remove(lastUsed);}}final C conn = this.connFactory.create(route);entry = pool.add(conn);this.leased.add(entry);return entry;}}boolean success = false;try {pool.queue(future);this.pending.add(future);if (deadline != null) {success = this.condition.awaitUntil(deadline);} else {this.condition.await();success = true;}if (future.isCancelled()) {throw new ExecutionException(operationAborted());}} finally {// In case of 'success', we were woken up by the// connection pool and should now have a connection// waiting for us, or else we're shutting down.// Just continue in the loop, both cases are checked.pool.unqueue(future);this.pending.remove(future);}// check for spurious wakeup vs. timeoutif (!success && (deadline != null && deadline.getTime() <= System.currentTimeMillis())) {break;}}throw new TimeoutException("Timeout waiting for connection");} finally {this.lock.unlock();}}

getPoolEntryBlocking先根据route从routeToPool取出对应的RouteSpecificPool，然后pool.getFree(state)，之后判断是否过期，是否关闭，没问题则从available移除，添加到leased中，然后执行onReuse回调，如果entry为null则通过connFactory.create(route)来创建

release

    @Overridepublic void release(final E entry, final boolean reusable) {this.lock.lock();try {if (this.leased.remove(entry)) {final RouteSpecificPool<T, C, E> pool = getPool(entry.getRoute());pool.free(entry, reusable);if (reusable && !this.isShutDown) {this.available.addFirst(entry);} else {entry.close();}onRelease(entry);Future<E> future = pool.nextPending();if (future != null) {this.pending.remove(future);} else {future = this.pending.poll();}if (future != null) {this.condition.signalAll();}}} finally {this.lock.unlock();}}

release方法先获取RouteSpecificPool，然后执行pool.free(entry, reusable)

CPool

org/apache/http/impl/conn/CPool.java

@Contract(threading = ThreadingBehavior.SAFE)
class CPool extends AbstractConnPool<HttpRoute, ManagedHttpClientConnection, CPoolEntry> {private static final AtomicLong COUNTER = new AtomicLong();private final Log log = LogFactory.getLog(CPool.class);private final long timeToLive;private final TimeUnit timeUnit;public CPool(final ConnFactory<HttpRoute, ManagedHttpClientConnection> connFactory,final int defaultMaxPerRoute, final int maxTotal,final long timeToLive, final TimeUnit timeUnit) {super(connFactory, defaultMaxPerRoute, maxTotal);this.timeToLive = timeToLive;this.timeUnit = timeUnit;}@Overrideprotected CPoolEntry createEntry(final HttpRoute route, final ManagedHttpClientConnection conn) {final String id = Long.toString(COUNTER.getAndIncrement());return new CPoolEntry(this.log, id, route, conn, this.timeToLive, this.timeUnit);}@Overrideprotected boolean validate(final CPoolEntry entry) {return !entry.getConnection().isStale();}@Overrideprotected void enumAvailable(final PoolEntryCallback<HttpRoute, ManagedHttpClientConnection> callback) {super.enumAvailable(callback);}@Overrideprotected void enumLeased(final PoolEntryCallback<HttpRoute, ManagedHttpClientConnection> callback) {super.enumLeased(callback);}}

CPool继承了AbstractConnPool，其T为HttpRoute，C为ManagedHttpClientConnection，E为CPoolEntry；其createEntry方法创建CPoolEntry，validate则判断connect是不是stale

小结

ConnPool定义了lease及release方法，其中定义了两个泛型，T表示route，E表示poolEntry；AbstractConnPool声明实现ConnPool、ConnPoolControl接口，它定义E必须继承PoolEntry，同时定义了泛型C，表示connectionType；CPool继承了AbstractConnPool，其T为HttpRoute，C为ManagedHttpClientConnection，E为CPoolEntry。

AbstractConnPool的lease方法主要是get及cancel，其中get方法主要是执行getPoolEntryBlocking，对于validateAfterInactivity大于0的则根据判断是否需要validate，若需要且validate失败则执行leasedEntry.close()及release方法；release方法先获取RouteSpecificPool，然后执行pool.free(entry, reusable)

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.xdnf.cn/news/149062.html

如若内容造成侵权/违法违规/事实不符，请联系一条长河网进行投诉反馈，一经查实，立即删除！

相关文章

《幸福之路》罗素（读书笔记）

《幸福之路》罗素（读书笔记）

目录 作者简介 作者的感悟 经典摘录 一、不幸福的成因 1、一部分要归咎于社会制度 2、一部分则得归咎于个人心理——当然&#xff0c;你可以说个人心理是社会制度的产物。 二、欠缺某些想要的东西&#xff0c;是快乐的必要条件 三、无聊与刺激 四、现代人的精神疲劳 五…
阅读更多...
【C++】vector相关OJ

【C++】vector相关OJ

文章目录 1. 只出现一次的数字2. 杨辉三角3. 电话号码字母组合 ヾ(๑╹◡╹)&#xff89;" 人总要为过去的懒惰而付出代价ヾ(๑╹◡╹)&#xff89;" 1. 只出现一次的数字 力扣链接 代码展示&#xff1a; class Solution { public:int singleNumber(vector<i…
阅读更多...
【“栈、队列”的应用】408数据结构代码

【“栈、队列”的应用】408数据结构代码

王道数据结构强化课——【“栈、队列”的应用】代码&#xff0c;持续更新 链式存储栈&#xff08;单链表实现&#xff09;&#xff0c;并基于上述定义&#xff0c;栈顶在链头&#xff0c;实现“出栈、入栈、判空、判满”四个基本操作 #include <stdio.h> #include <…
阅读更多...
大数据-玩转数据-Flink SQL编程实战 (热门商品TOP N)

大数据-玩转数据-Flink SQL编程实战 (热门商品TOP N)

一、需求描述 每隔30min 统计最近 1hour的热门商品 top3, 并把统计的结果写入到mysql中。 二、需求分析 1.统计每个商品的点击量, 开窗2.分组窗口分组3.over窗口 三、需求实现 3.1、创建数据源示例 input/UserBehavior.csv 543462,1715,1464116,pv,1511658000 662867,22…
阅读更多...
基于阶梯碳交易的含P2G-CCS耦合和燃气掺氢的虚拟电厂优化调度（matlab代码）

基于阶梯碳交易的含P2G-CCS耦合和燃气掺氢的虚拟电厂优化调度（matlab代码）

目录 1 主要内容 系统结构图 P2G-CCS 耦合模型 其他算例对比 2 部分代码 3 下载链接 1 主要内容 该程序复现《基于阶梯碳交易的含P2G-CCS耦合和燃气掺氢的虚拟电厂优化调度》模型&#xff0c;以碳交易和碳封存成本、燃煤机组启停和煤耗成本、弃风成本、购气成本之和为目标…
阅读更多...
vertx的学习总结6

vertx的学习总结6

Beyond the event bus 一、章节覆盖&#xff1a; 如何在事件总线之上公开服务 verticles和事件总线服务的异步测试 动态代理&#xff1a; MyService 接口 package porxy.test;import io.vertx.codegen.annotations.ProxyGen;ProxyGen public interface MyService {void he…
阅读更多...
智慧公厕：城市公共厕所的未来之路

智慧公厕：城市公共厕所的未来之路

随着城市化进程的不断推进&#xff0c;人们对城市环境质量的要求也越来越高。在城市管理中&#xff0c;公厕作为一个必不可少的公共设施&#xff0c;不仅关乎城市的文明形象&#xff0c;还与市民的生活质量密切相关。为了解决传统公厕存在的问题&#xff0c;智慧公厕应运而生。…
阅读更多...
Go-Python-Java-C-LeetCode高分解法-第八周合集

Go-Python-Java-C-LeetCode高分解法-第八周合集

前言 本题解Go语言部分基于 LeetCode-Go 其他部分基于本人实践学习 个人题解GitHub连接&#xff1a;LeetCode-Go-Python-Java-C 欢迎订阅CSDN专栏&#xff0c;每日一题&#xff0c;和博主一起进步 LeetCode专栏 我搜集到了50道精选题&#xff0c;适合速成概览大部分常用算法 突…
阅读更多...
大模型部署手记（3）通义千问+Windows GPU

大模型部署手记（3）通义千问+Windows GPU

1.简介 组织机构&#xff1a;阿里 代码仓&#xff1a;GitHub - QwenLM/Qwen: The official repo of Qwen (通义千问) chat & pretrained large language model proposed by Alibaba Cloud. 模型&#xff1a;Qwen/Qwen-7B-Chat-Int4 下载&#xff1a;http://huggingface…
阅读更多...
【AI视野·今日Sound 声学论文速览 第十八期】Wed, 4 Oct 2023

【AI视野·今日Sound 声学论文速览 第十八期】Wed, 4 Oct 2023

AI视野今日CS.Sound 声学论文速览 Wed, 4 Oct 2023 Totally 4 papers &#x1f449;上期速览✈更多精彩请移步主页 Daily Sound Papers Mel-Band RoFormer for Music Source Separation Authors Ju Chiang Wang, Wei Tsung Lu, Minz Won最近&#xff0c;基于多频段频谱图的方法…
阅读更多...
windows server 2012 服务器打开系统远程功能

windows server 2012 服务器打开系统远程功能

服务器上开启远程功能 进入服务器&#xff0c;选择“添加角色和功能” 需要选择安装的服务器类型&#xff0c;如图所示 然后在服务器池中选择你需要使用的服务器。 选择完成后&#xff0c;在图示列表下勾选“远程桌面服务” 再选择需要安装的功能和角色服务。 选择完成确认内容…
阅读更多...
大模型部署手记（4）MOSS+Jetson AGX Orin

大模型部署手记（4）MOSS+Jetson AGX Orin

1.简介 组织机构&#xff1a;复旦大学 代码仓&#xff1a;GitHub - OpenLMLab/MOSS: An open-source tool-augmented conversational language model from Fudan University 模型&#xff1a;fnlp/moss-moon-003-sft-int4 下载&#xff1a;https://huggingface.co/fnlp/mos…
阅读更多...
C++_pen_友元

C++_pen_友元

友元&#xff08;破坏封装&#xff09; 我故意让别人能使用我的私有成员 友元类 friend class B;友元函数 friend void func();友元成员函数 friend void A::func();例 #include <stdio.h>class A;class C{ public:void CprintA(A &c); };class B{ public:void Bpri…
阅读更多...
qt 5.15.2 安卓 macos

qt 5.15.2 安卓 macos

macos环境安卓配置 我的系统是monterey12.5.1 打开qt的配置界面 这里版本是java1.8&#xff0c;注意修改这个json文件&#xff0c;显示包内容 {"common": {"sdk_tools_url": {"linux": "https://dl.google.com/android/repository/comm…
阅读更多...
lv7 嵌入式开发-网络编程开发 07 TCP服务器实现

lv7 嵌入式开发-网络编程开发 07 TCP服务器实现

目录 1 函数介绍 1.1 socket函数 与 通信域 1.2 bind函数 与 通信结构体 1.3 listen函数 与 accept函数 2 TCP服务端代码实现 3 TCP客户端代码实现 4 代码优化 5 练习 1 函数介绍 其中read、write、close在IO中已经介绍过&#xff0c;只需了解socket、bind、listen、acc…
阅读更多...
Python爬虫案例入门教程（纯小白向）——夜读书屋小说

Python爬虫案例入门教程（纯小白向）——夜读书屋小说

Python爬虫案例——夜读书屋小说 前言 如果你是python小白并且对爬虫有着浓厚的兴趣&#xff0c;但是面对网上错综复杂的实战案例看也看不懂&#xff0c;那么你可以尝试阅读我的文章&#xff0c;我也是从零基础python开始学习爬虫&#xff0c;非常清楚在过程中所遇到的困难&am…
阅读更多...
简单查找重复文本文件

简单查找重复文本文件

声明这是最初 我的提问给个文本分类清单input查找文件夹下 .py .txt .excel .word 一模一样的文本不是找文件名 找相同格式下的文件文本是否一样 文件单独复制到文件夹下两个文件全部复制到文件夹下 print 打印相同文本文件的名字 比如查找到了3.py与4.5.是.py文件中的文本文件…
阅读更多...
Scala第一章节

Scala第一章节

Scala第一章节 scala总目录 章节目标 理解Scala的相关概述掌握Scala的环境搭建掌握Scala小案例: 做最好的自己 1. Scala简介 1.1 概述 ​ Scala(斯嘎拉)这个名字来源于"Scalable Language(可伸缩的语言)", 它是一门基于JVM的多范式编程语言, 通俗的说: Scala是一…
阅读更多...
JAVAWeb业务层开发-＞普通和基于MP

JAVAWeb业务层开发-＞普通和基于MP

普通方式业务层开发 service定义接口&#xff08;主要实现逻辑层面的业务功能&#xff09; serviceImpl实现该接口 注意事项&#xff1a; 逻辑判断的代码可以使用&#xff1e;号&#xff0c;使得返回结果为布尔类型。 小结&#xff1a;每一个接口写完都要写测试类去检测&#…
阅读更多...
JMeter的详细使用及相关问题

JMeter的详细使用及相关问题

一、中文乱码问题 如果出现乱码&#xff0c;需要修改编码集&#xff0c;&#xff08;版本问题有的不需要修改&#xff0c;就不用管&#xff09; 修改后保存重启就好了。 JMeter5.5版本的按照如下修改&#xff1a; 二、JMeter的启动 ①建议直接用ApacheJMeter.jar双击启动…
阅读更多...
最新文章

Copyright @ 2022~2023