"Redis客户端连接数一直降不下来"的有关问题解决 good

2023-08-15 09:28:04

[线上问题] "Redis客户端连接数一直降不下来"的问题解决

前段时间，上线了新的 Redis缓存（Cache）服务，准备替换掉 Memcached。

为什么要将 Memcached 替换掉？

原因是 业务数据是压缩后的列表型数据，缓存中保存最新的3000条数据。对于新数据追加操作，需要拆解成[get + unzip + append + zip + set]这5步操作。若列表长度在O(1k)级别的，其耗时至少在50ms+。而在并发环境下，这样会存在“数据更新覆盖问题”，因为追加操作不是原子操作。（线上也确实遇到了这个问题）

针对“追加操作不是原子操作”的问题，我们就开始调研有哪些可以解决这个问题同时又满足业务数据类型的分布式缓存解决方案。

当前，业界常用的一些 key-value分布式缓存系统如下：

  • Redis
  • Memcached
  • Cassandra
  • Tokyo Tyrant (Tokyo Cabinet)

参考自：

  • 2010年的技术架构建议 – Tim Yang
  • From distributed caches to in-memory data grids
  • Cassandra vs MongoDB vs CouchDB vs Redis vs Riak vs HBase vs Couchbase vs OrientDB vs Aerospike vs Hypertable vs ElasticSearch vs Accumulo vs VoltDB vs Scalaris comparison

通过对比、筛选分析，我们最终选择了 Redis。原因有以下几个：

  • Redis 是一个 key-value 的缓存(cache)和存储(store)系统（现在我们只用它来做缓存，目前还未当作DB用，数据存放在 Cassandra 里）
  • 支持丰富的数据结构，List 就专门用于存储列表型数据，默认按操作时间排序。Sorted Set 可以按分数排序元素，分数是一种广义概念，可以是时间或评分。其次，其丰富的数据结构为日后扩展提供了很大的方便。
  • 提供的所有操作都是原子操作，为并发天然保驾护航。
  • 超快的性能，见其官方性能测试《How fast is Redis?》。
  • 拥有比较成熟的Java客户端 - Jedis，像新浪微博都是使用它作为客户端。（官方推荐的Clients）

啰嗦了一些其它东西，现在言归正传。

Redis 服务上线当天，就密切关注 Redis 的一些重要监控指标（clients：客户端连接数、memory、stats：服务器每秒钟执行的命令数量、commandstats：一些关键命令的执行统计信息、redis.error.log：异常日志）。（参考自《Redis监控方案》）

观察到下午5点左右，发现“客户端连接数”一直在增长，最高时都超过了2000个（见下图），即使减少也就减1~2个。但应用的QPS却在 10 个左右，而线上应用服务器不超过10台。按理说，服务器肯定不会有这么高的连接数，肯定哪里使用有问题。

"Redis客户端连接数一直降不下来"的有关问题解决 good

现在只能通过逆向思维反向来推测问题：

  • Redis服务端监控到的“客户端连接数”表明所有客户端总和起来应该有那么多，所以首先到各个应用服务器上确认连接数量；
  • 通过“sudo netstat -antp | grep 6379 | wc -l”确认，有一台应用Redis的连接数都超过了1000个，另一台应用则在400左右，其它的都在60上下。（60上下是正常的）
  • 第一个问题：为什么不同的机器部署了同一个应用程序，表现出来的行为却是不一样？
  • 第二个问题：连接数超过1000个的那台，其请求量(140)是比其它机器(200+)要低的(因为它在Nginx中配置的权重低)，那它的连接数为什么会这么高？到底发生了什么？
  • 对于“第二个问题”，我们通过各个应用的Redis异常日志（redis.error.log）知道发生了什么。最高那台应用的异常操作特别多，共有130+个异常，且存在“关闭集群链接时异常导致连接泄漏”问题；另一台较高的应用也存在类似的情况，而其它正常的应用则不超过2个异常，且不存在“连接泄漏”问题。这样，“第二个问题”算是弄清楚了。（“连接泄漏”问题具体如何修复见《[FAQ] Jedis使用过程中踩过的那些坑》）
  • 至此，感觉问题好像已经解决了，但其实没有。通过连续几天的观察，发现最高的时候，它的连接数甚至超过了3000+，这太恐怖了。（当时 leader 还和我说，要不要重启一下应用）
  • 即使应用的QPS是 20个/s，且存在“连接泄漏”问题，连接数也不会超过1000+。但现在连接数居然达到了3000+，这说不通，只有一个可能就是未正确使用Jedis。
  • 这时候就继续反推，Redis的连接数反映了Jedis对象池的池对象数量。线上部署了2台Redis服务器作为一个集群，说明这台应用共持有(3000/2=1500)个池对象。(因为Jedis基于Apache Commons Pool的GenericObjectPool实现)
  • 第三个问题：根据应用的QPS，每秒钟请求需要的Active池对象也不会超过20个，那其余的1480个都是“空闲池对象”。为什么那么多的“空闲池对象”未被释放？
  • 现在就来反思：Jedis的那些配置属性与对象池管理“空闲池对象”相关，GenericObjectPool背后是怎么管理“空闲池对象”的？

由于在使用Jedis的过程中，就对Apache Commons Pool摸了一次底。对最后的两个疑惑都比较了解，Jedis的以下这些配置与对象池管理“空闲池对象”相关：

redis.max.idle.num=32768
redis.min.idle.num=30
redis.pool.behaviour=FIFO
redis.time.between.eviction.runs.seconds=1
redis.num.tests.per.eviction.run=10
redis.min.evictable.idle.time.minutes=5
redis.max.evictable.idle.time.minutes=1440

在上面说“每台应用的Jedis连接数在60个左右是正常的”的理由是：线上共部署了2台Redis服务器，Jedis的“最小空闲池对象个数”配置为30 (redis.min.idle.num=30)。

GenericObjectPool是通过“驱逐者线程Evictor”管理“空闲池对象”的，详见《Apache Commons Pool之空闲对象的驱逐检测机制》一文。最下方的5个配置都是与“驱逐者线程Evictor”相关的，表示对象池的空闲队列行为为FIFO“先进先出”队列方式，每秒钟(1)检测10个空闲池对象，空闲池对象的空闲时间只有超过5分钟后，才有资格被驱逐检测，若空闲时间超过一天(1440)，将被强制驱逐。

因为“驱逐者线程Evictor”会无限制循环地对“池对象空闲队列”进行迭代式地驱逐检测。空闲队列的行为有两种方式：LIFO“后进先出”栈方式、FIFO“先进先出”队列方式，默认使用LIFO。下面通过两幅图来展示这两种方式的实际运作方式：

一、LIFO“后进先出”栈方式

"Redis客户端连接数一直降不下来"的有关问题解决 good

二、FIFO“先进先出”队列方式

"Redis客户端连接数一直降不下来"的有关问题解决 good

从上面这两幅图可以看出，LIFO“后进先出”栈方式 有效地利用了空闲队列里的热点池对象资源，随着流量的下降会使一些池对象长时间未被使用而空闲着，最终它们将被淘汰驱逐；
而 FIFO“先进先出”队列方式 虽然使空闲队列里所有池对象都能在一段时间里被使用，看起来它好像分散了资源的请求，但其实这不利于资源的释放(因为空闲池对象的空闲时间只有超过5分钟后，才有资格被驱逐检测，分散资源请求的同时，也导致符合释放条件的空闲对象也变少了，而每个空闲对象都占用一个redis连接)。
而这也是“客户端连接数一直降不下来”的根源之一。

redis.pool.behaviour=FIFO
redis.time.between.eviction.runs.seconds=1
redis.num.tests.per.eviction.run=10
redis.min.evictable.idle.time.minutes=5

按照上述配置，我们可以计算一下，5分钟里到底有多少个空闲池对象被循环地使用过。
根据应用QPS 10个/s计算，5分钟里大概有10*5*60=3000个空闲池对象被使用过，正好与上面的“连接数尽然达到了3000+”符合，这样就说得通了。至此，整个问题终于水落石出了。（从监控图也可以看出，在21号晚上6点左右修改配置重启服务后，连接数就比较平稳了）

这里还要解释一下为什么使用FIFO“先进先出”队列方式的空闲队列行为？

因为我们在Jedis的基础上开发了“故障节点自动摘除，恢复正常的节点自动添加”的功能，本来想使用FIFO“先进先出”队列方式在节点故障时，对象池能快速更新整个集群信息，没想到弄巧成拙了。

修复后的Jedis配置如下：

redis.max.idle.num=32768
redis.min.idle.num=30
redis.pool.behaviour=LIFO
redis.time.between.eviction.runs.seconds=1
redis.num.tests.per.eviction.run=10
redis.min.evictable.idle.time.minutes=5
redis.max.evictable.idle.time.minutes=30

综上所述，这个问题发生有两方面的原因：

    1. 未正确使用对象池的空闲队列行为（LIFO“后进先出”栈方式）
    2. “关闭集群链接时异常导致连接泄漏”问题

http://www.myexception.cn/internet/1849994.html

本文主要剖析 Apache Commons Pool 的“空闲对象的驱逐检测机制”的实现原理。

以下面3个步骤来循序渐进地深入剖析其实现原理：

  1. 启动“空闲对象的驱逐者线程”（startEvictor(...)）的2个入口
  2. 在启动时，创建一个新的"驱逐者线程"(Evictor)，并使用"驱逐者定时器"(EvictionTimer)进行调度
  3. 进入真正地"空闲池对象"的驱逐检测操作（evict()）

下图是“空闲对象的驱逐检测机制”处理流程的时序图(阅读代码时结合着看可以加深理解)：

"Redis客户端连接数一直降不下来"的有关问题解决 good

GenericObjectPool.evict() 处理流程的时序图：

"Redis客户端连接数一直降不下来"的有关问题解决 good

GenericObjectPool.ensureMinIdle()处理流程的时序图：

"Redis客户端连接数一直降不下来"的有关问题解决 good

一、启动“空闲对象的驱逐者线程”（startEvictor(...)）共有2个入口

1. GenericObjectPool 构造方法

GenericObjectPool(...)：初始化"池对象工厂"，设置"对象池配置"，并启动"驱逐者线程"。

  1. /**
  2. * 使用特定的配置来创建一个新的"通用对象池"实例。
  3. *
  4. * @param factory   The object factory to be used to create object instances
  5. *                  used by this pool (用于创建池对象实例的对象工厂)
  6. * @param config    The configuration to use for this pool instance. (用于该对象池实例的配置信息)
  7. *                  The configuration is used by value. Subsequent changes to
  8. *                  the configuration object will not be reflected in the
  9. *                  pool. (随后对配置对象的更改将不会反映到池中)
  10. */
  11. public GenericObjectPool(PooledObjectFactory<T> factory,
  12. GenericObjectPoolConfig config) {
  13. super(config, ONAME_BASE, config.getJmxNamePrefix());
  14. if (factory == null) {
  15. jmxUnregister(); // tidy up
  16. throw new IllegalArgumentException("factory may not be null");
  17. }
  18. this.factory = factory;
  19. this.setConfig(config);
  20. // 启动"驱逐者线程"
  21. startEvictor(this.getTimeBetweenEvictionRunsMillis());
  22. }

2. BaseGenericObjectPool.setTimeBetweenEvictionRunsMillis(...) - 设置"驱逐者线程"的运行间隔时间

可以动态地更新"驱逐者线程"的运行调度间隔时间。

  1. /**
  2. * 设置"空闲对象的驱逐者线程"的运行调度间隔时间。（同时，会立即启动"驱逐者线程"）
  3. * <p>
  4. * 如果该值是非正数，则没有"空闲对象的驱逐者线程"将运行。
  5. * <p>
  6. * 默认是 {@code -1}，即没有"空闲对象的驱逐者线程"在后台运行着。
  7. * <p>
  8. * 上一层入口：{@link GenericObjectPool#setConfig(GenericObjectPoolConfig)}<br>
  9. * 顶层入口：{@link GenericObjectPool#GenericObjectPool(PooledObjectFactory, GenericObjectPoolConfig)}，
  10. * 在最后还会调用{@link #startEvictor(long)}来再次启动"空闲对象的驱逐者线程"。<br>
  11. * 这样在初始化时，这里创建的"驱逐者线程"就多余了，会立刻被销毁掉。<br>
  12. * 但这里为什么要这样实现呢？<br>
  13. * 我的理解是：为了能动态地更新"驱逐者线程"的调度间隔时间。
  14. *
  15. * @param timeBetweenEvictionRunsMillis
  16. *            number of milliseconds to sleep between evictor runs ("驱逐者线程"运行的间隔毫秒数)
  17. *
  18. * @see #getTimeBetweenEvictionRunsMillis
  19. */
  20. public final void setTimeBetweenEvictionRunsMillis(
  21. long timeBetweenEvictionRunsMillis) {
  22. this.timeBetweenEvictionRunsMillis = timeBetweenEvictionRunsMillis;
  23. // 启动"驱逐者线程"
  24. this.startEvictor(timeBetweenEvictionRunsMillis);
  25. }

二、startEvictor(long delay)- 启动“空闲对象的驱逐者线程”

如果有一个"驱逐者线程"(Evictor)运行着，则会先停止它；

然后创建一个新的"驱逐者线程"，并使用"驱逐者定时器"(EvictionTimer)进行调度。

  1. // 空闲对象的驱逐回收策略
  2. /** 用于初始化"驱逐者线程"的同步对象 */
  3. final Object evictionLock = new Object();
  4. /** 空闲对象驱逐者线程 */
  5. private Evictor evictor = null; // @GuardedBy("evictionLock")
  6. /** 驱逐检测对象迭代器 */
  7. Iterator<PooledObject<T>> evictionIterator = null; // @GuardedBy("evictionLock")
  8. /**
  9. * 启动"空闲对象的驱逐者线程"。
  10. * <p>
  11. * 如果有一个"驱逐者线程"({@link Evictor})运行着，则会先停止它；
  12. * 然后创建一个新的"驱逐者线程"，并使用"驱逐者定时器"({@link EvictionTimer})进行调度。
  13. *
  14. * <p>This method needs to be final, since it is called from a constructor. (因为它被一个构造器调用)
  15. * See POOL-195.</p>
  16. *
  17. * @param delay time in milliseconds before start and between eviction runs (驱逐者线程运行的开始和间隔时间 毫秒数)
  18. */
  19. final void startEvictor(long delay) {
  20. synchronized (evictionLock) { // 同步锁
  21. if (null != evictor) {
  22. // 先释放申请的资源
  23. EvictionTimer.cancel(evictor);
  24. evictor = null;
  25. evictionIterator = null;
  26. }
  27. if (delay > 0) {
  28. evictor = new Evictor();
  29. EvictionTimer.schedule(evictor, delay, delay);
  30. }
  31. }
  32. }

2.1 Evictor - "驱逐者线程"实现

Evictor，"空闲对象的驱逐者"定时任务，继承自 TimerTask。TimerTask 是一个可由定时器（Timer）调度执行一次或重复执行的任务。

核心实现逻辑：

1. evict()：执行numTests个空闲池对象的驱逐测试，驱逐那些符合驱逐条件的被检测对象；
2. ensureMinIdle()：试图确保配置的对象池中可用"空闲池对象"实例的最小数量。

  1. /**
  2. * Class loader for evictor thread to use since in a J2EE or similar
  3. * environment the context class loader for the evictor thread may have
  4. * visibility of the correct factory. See POOL-161.
  5. * 驱逐者线程的类加载器
  6. */
  7. private final ClassLoader factoryClassLoader;
  8. // Inner classes
  9. /**
  10. * "空闲对象的驱逐者"定时任务，继承自{@link TimerTask}。
  11. *
  12. * @see GenericObjectPool#GenericObjectPool(PooledObjectFactory, GenericObjectPoolConfig)
  13. * @see GenericKeyedObjectPool#setTimeBetweenEvictionRunsMillis(long)
  14. */
  15. class Evictor extends TimerTask {
  16. /**
  17. * 运行对象池维护线程。
  18. * 驱逐对象具有驱逐者的资格，同时保证空闲实例可用的最小数量。
  19. * 因为调用"驱逐者线程"的定时器是被所有对象池共享的，
  20. * 但对象池可能存在不同的类加载器中，所以驱逐者必须确保采取的任何行为
  21. * 都得在与对象池相关的工厂的类加载器下。
  22. */
  23. @Override
  24. public void run() {
  25. ClassLoader savedClassLoader =
  26. Thread.currentThread().getContextClassLoader();
  27. try {
  28. // Set the class loader for the factory (设置"工厂的类加载器")
  29. Thread.currentThread().setContextClassLoader(
  30. factoryClassLoader);
  31. // Evict from the pool (从"对象池"中驱逐)
  32. try {
  33. // 1. 执行numTests个空闲池对象的驱逐测试，驱逐那些符合驱逐条件的被检测对象
  34. evict(); // 抽象方法
  35. } catch(Exception e) {
  36. swallowException(e);
  37. } catch(OutOfMemoryError oome) {
  38. // Log problem but give evictor thread a chance to continue
  39. // in case error is recoverable
  40. oome.printStackTrace(System.err);
  41. }
  42. // Re-create idle instances. (重新创建"空闲池对象"实例)
  43. try {
  44. // 2. 试图确保配置的对象池中可用"空闲池对象"实例的最小数量
  45. ensureMinIdle(); // 抽象方法
  46. } catch (Exception e) {
  47. swallowException(e);
  48. }
  49. } finally {
  50. // Restore the previous CCL
  51. Thread.currentThread().setContextClassLoader(savedClassLoader);
  52. }
  53. }
  54. }

2.2 EvictionTimer - "驱逐者定时器"实现

EvictionTimer，提供一个所有"对象池"共享的"空闲对象的驱逐定时器"。此类包装标准的定时器(Timer)，并追踪有多少个"对象池"使用它。

核心实现逻辑：

schedule(TimerTask task, long delay, long period)：添加指定的驱逐任务到这个定时器

  1. /**
  2. * 提供一个所有"对象池"共享的"空闲对象的驱逐定时器"。
  3. *
  4. * 此类包装标准的定时器({@link Timer})，并追踪有多少个对象池使用它。
  5. *
  6. * 如果没有对象池使用这个定时器，它会被取消。这样可以防止线程一直运行着
  7. * (这会导致内存泄漏)，防止应用程序关闭或重新加载。
  8. * <p>
  9. * 此类是包范围的，以防止其被纳入到池框架的公共API中。
  10. * <p>
  11. * <font color="red">此类是线程安全的！</font>
  12. *
  13. * @since 2.0
  14. */
  15. class EvictionTimer {
  16. /** Timer instance (定时器实例) */
  17. private static Timer _timer; //@GuardedBy("this")
  18. /** Static usage count tracker (使用计数追踪器) */
  19. private static int _usageCount; //@GuardedBy("this")
  20. /** Prevent instantiation (防止实例化) */
  21. private EvictionTimer() {
  22. // Hide the default constructor
  23. }
  24. /**
  25. * 添加指定的驱逐任务到这个定时器。
  26. * 任务，通过调用该方法添加的，必须调用{@link #cancel(TimerTask)}来取消这个任务，
  27. * 以防止内存或消除泄漏。
  28. *
  29. * @param task      Task to be scheduled (定时调度的任务)
  30. * @param delay     Delay in milliseconds before task is executed (任务执行前的等待时间)
  31. * @param period    Time in milliseconds between executions (执行间隔时间)
  32. */
  33. static synchronized void schedule(TimerTask task, long delay, long period) {
  34. if (null == _timer) {
  35. // Force the new Timer thread to be created with a context class
  36. // loader set to the class loader that loaded this library
  37. ClassLoader ccl = AccessController.doPrivileged(
  38. new PrivilegedGetTccl());
  39. try {
  40. AccessController.doPrivileged(new PrivilegedSetTccl(
  41. EvictionTimer.class.getClassLoader()));
  42. _timer = new Timer("commons-pool-EvictionTimer", true);
  43. } finally {
  44. AccessController.doPrivileged(new PrivilegedSetTccl(ccl));
  45. }
  46. }
  47. // 增加"使用计数器"，并调度"任务"
  48. _usageCount++;
  49. _timer.schedule(task, delay, period);
  50. }
  51. /**
  52. * 从定时器中删除指定的驱逐者任务。
  53. * <p>
  54. * Remove the specified eviction task from the timer.
  55. *
  56. * @param task      Task to be scheduled (定时调度任务)
  57. */
  58. static synchronized void cancel(TimerTask task) {
  59. task.cancel(); // 1. 将任务的状态标记为"取消(CANCELLED)"状态
  60. _usageCount--;
  61. if (_usageCount == 0) { // 2. 如果没有对象池使用这个定时器，定时器就会被取消
  62. _timer.cancel();
  63. _timer = null;
  64. }
  65. }
  66. /**
  67. * {@link PrivilegedAction} used to get the ContextClassLoader (获取"上下文类加载器")
  68. */
  69. private static class PrivilegedGetTccl implements PrivilegedAction<ClassLoader> {
  70. @Override
  71. public ClassLoader run() {
  72. return Thread.currentThread().getContextClassLoader();
  73. }
  74. }
  75. /**
  76. * {@link PrivilegedAction} used to set the ContextClassLoader (设置"上下文类加载器")
  77. */
  78. private static class PrivilegedSetTccl implements PrivilegedAction<Void> {
  79. /** ClassLoader */
  80. private final ClassLoader cl;
  81. /**
  82. * Create a new PrivilegedSetTccl using the given classloader
  83. * @param cl ClassLoader to use
  84. */
  85. PrivilegedSetTccl(ClassLoader cl) {
  86. this.cl = cl;
  87. }
  88. @Override
  89. public Void run() {
  90. Thread.currentThread().setContextClassLoader(cl);
  91. return null;
  92. }
  93. }
  94. }

三、"驱逐者线程"和"驱逐者定时器"都准备就绪，现在真正地开始"空闲池对象"的驱逐检测操作（evict()）

BaseGenericObjectPool.evict()：驱逐检测操作的抽象声明

  1. /**
  2. * 执行{@link numTests}个空闲池对象的驱逐测试，驱逐那些符合驱逐条件的被检测对象。
  3. * <p>
  4. * 如果{@code testWhileIdle}为{@code true}，则被检测的对象在访问期间是有效的(无效则会被删除)；
  5. * 否则，仅有那些池对象的空闲时间超过{@code minEvicableIdleTimeMillis}会被删除。
  6. *
  7. * @throws Exception when there is a problem evicting idle objects. (当这是一个有问题的驱逐空闲池对象时，才会抛出Exception异常。)
  8. */
  9. public abstract void evict() throws Exception;

GenericObjectPool.evict()："通用对象池"的驱逐检测操作实现

核心实现逻辑：

1. 确保"对象池"还打开着

2. 获取"驱逐回收策略"

3. 获取"驱逐配置"

4. 对所有待检测的"空闲对象"进行驱逐检测

4.1 初始化"驱逐检测对象(空闲池对象)的迭代器"

4.2 将"池对象"标记为"开始驱逐状态"

4.3 进行真正的"驱逐检测"操作（EvictionPolicy.evict(...)）

4.3.1 如果"池对象"是可驱逐的，则销毁它

4.3.2 否则，是否允许空闲时进行有效性测试

4.3.2.1 先激活"池对象"

4.3.2.2 使用PooledObjectFactory.validateObject(PooledObject)进行"池对象"的有效性校验

4.3.2.2.1 如果"池对象"不是有效的，则销毁它

4.3.2.2.2 否则，还原"池对象"状态

4.3.2.3 通知"空闲对象队列"，驱逐测试已经结束

5. 是否要移除"被废弃的池对象"

  1. /** 池的空闲池对象列表 */
  2. private final LinkedBlockingDeque<PooledObject<T>> idleObjects =
  3. new LinkedBlockingDeque<PooledObject<T>>();
  4. /** 池对象工厂 */
  5. private final PooledObjectFactory<T> factory;
  6. // 空闲对象的驱逐回收策略
  7. /** 用于初始化"驱逐者线程"的同步对象 */
  8. final Object evictionLock = new Object();
  9. /** 空闲对象驱逐者线程 */
  10. private Evictor evictor = null; // @GuardedBy("evictionLock")
  11. /** 驱逐检测对象("空闲池对象")的迭代器 */
  12. Iterator<PooledObject<T>> evictionIterator = null; // @GuardedBy("evictionLock")
  13. /** 被废弃的池对象追踪的配置属性 */
  14. private volatile AbandonedConfig abandonedConfig = null;
  15. /**
  16. * {@inheritDoc}
  17. * <p>
  18. * 按顺序对被审查的对象进行连续驱逐检测，对象是以"从最老到最年轻"的顺序循环。
  19. */
  20. @Override
  21. public void evict() throws Exception {
  22. // 1. 确保"对象池"还打开着
  23. this.assertOpen();
  24. if (idleObjects.size() > 0) {
  25. PooledObject<T> underTest = null; // 测试中的池对象
  26. // 2. 获取"驱逐回收策略"
  27. EvictionPolicy<T> evictionPolicy = this.getEvictionPolicy();
  28. synchronized (evictionLock) { // 驱逐锁定
  29. // 3. 获取"驱逐配置"
  30. EvictionConfig evictionConfig = new EvictionConfig(
  31. this.getMinEvictableIdleTimeMillis(),
  32. this.getSoftMinEvictableIdleTimeMillis(),
  33. this.getMinIdle()
  34. );
  35. // 4. 对所有待检测的"空闲对象"进行驱逐检测
  36. for (int i = 0, m = this.getNumTests(); i < m; i++) {
  37. // 4.1 初始化"驱逐检测对象(空闲池对象)的迭代器"
  38. if (evictionIterator == null || !evictionIterator.hasNext()) { // 已对所有空闲对象完成一次遍历
  39. // 根据"对象池使用行为"赋值驱逐迭代器
  40. if (this.getLifo()) {
  41. evictionIterator = idleObjects.descendingIterator();
  42. } else {
  43. evictionIterator = idleObjects.iterator();
  44. }
  45. }
  46. if (!evictionIterator.hasNext()) {
  47. // Pool exhausted, nothing to do here (对象池被耗尽，无可用池对象)
  48. return;
  49. }
  50. try {
  51. underTest = evictionIterator.next();
  52. } catch (NoSuchElementException nsee) {
  53. // Object was borrowed in another thread (池对象被其它请求线程借用了)
  54. // Don't count this as an eviction test so reduce i;
  55. i--;
  56. evictionIterator = null;
  57. continue;
  58. }
  59. // 4.2 将"池对象"标记为"开始驱逐状态"
  60. if (!underTest.startEvictionTest()) {
  61. // Object was borrowed in another thread
  62. // Don't count this as an eviction test so reduce i;
  63. i--;
  64. continue;
  65. }
  66. boolean testWhileIdle = this.getTestWhileIdle(); // 是否要在对象空闲时测试有效性
  67. // 4.3 进行真正的"驱逐检测"操作（EvictionPolicy.evict(...)）
  68. if (evictionPolicy.evict(evictionConfig, underTest,
  69. idleObjects.size())) {
  70. // 4.3.1 如果"池对象"是可驱逐的，则销毁它
  71. this.destroy(underTest);
  72. destroyedByEvictorCount.incrementAndGet();
  73. } else {
  74. // 4.3.2 否则，是否允许空闲时进行有效性测试
  75. if (testWhileIdle) { // 允许空闲时进行有效性测试
  76. // 4.3.2.1 先激活"池对象"
  77. boolean active = false;
  78. try {
  79. factory.activateObject(underTest);
  80. active = true;
  81. } catch (Exception e) {
  82. this.destroy(underTest);
  83. destroyedByEvictorCount.incrementAndGet();
  84. }
  85. // 4.3.2.2 使用PooledObjectFactory.validateObject(PooledObject)进行"池对象"的有效性校验
  86. if (active) {
  87. if (!factory.validateObject(underTest)) {
  88. // 4.3.2.2.1 如果"池对象"不是有效的，则销毁它
  89. this.destroy(underTest);
  90. destroyedByEvictorCount.incrementAndGet();
  91. } else {
  92. try {
  93. // 4.3.2.2.2 否则，还原"池对象"状态
  94. factory.passivateObject(underTest);
  95. } catch (Exception e) {
  96. this.destroy(underTest);
  97. destroyedByEvictorCount.incrementAndGet();
  98. }
  99. }
  100. }
  101. }
  102. // 4.3.2.3 通知"空闲对象队列"，驱逐测试已经结束
  103. if (!underTest.endEvictionTest(idleObjects)) {
  104. // TODO - May need to add code here once additional
  105. // states are used
  106. }
  107. }
  108. }
  109. }
  110. }
  111. // 5. 是否要移除"被废弃的池对象"
  112. AbandonedConfig ac = this.abandonedConfig;
  113. if (ac != null && ac.getRemoveAbandonedOnMaintenance()) {
  114. this.removeAbandoned(ac);
  115. }
  116. }

BaseGenericObjectPool.ensureMinIdle()："确保对象池中可用"空闲池对象"实例的最小数量"的抽象声明

  1. /**
  2. * 试图确保配置的对象池中可用"空闲池对象"实例的最小数量。
  3. *
  4. * @throws Exception if an error occurs creating idle instances
  5. */
  6. abstract void ensureMinIdle() throws Exception;

GenericObjectPool.ensureMinIdle()："确保对象池中可用"空闲池对象"实例的最小数量"实现

  1. @Override
  2. void ensureMinIdle() throws Exception {
  3. this.ensureIdle(this.getMinIdle(), true);
  4. }
  5. /**
  6. * 返回对象池中维护的空闲对象的最小数量目标。
  7. * <p>
  8. * 此设置仅会在{@link #getTimeBetweenEvictionRunsMillis()}的返回值大于0时，
  9. * 且该值是正整数时才会生效。
  10. * <p>
  11. * 默认是 {@code 0}，即对象池不维护空闲的池对象。
  12. *
  13. * @return The minimum number of objects. (空闲对象的最小数量)
  14. *
  15. * @see #setMinIdle(int)
  16. * @see #setMaxIdle(int)
  17. * @see #setTimeBetweenEvictionRunsMillis(long)
  18. */
  19. @Override
  20. public int getMinIdle() {
  21. int maxIdleSave = this.getMaxIdle();
  22. if (this.minIdle > maxIdleSave) {
  23. return maxIdleSave;
  24. } else {
  25. return minIdle;
  26. }
  27. }
  28. /**
  29. * 试图确保对象池中存在的{@code idleCount}个空闲实例。
  30. * <p>
  31. * 创建并添加空闲实例，直到空闲实例数量({@link #getNumIdle()})达到{@code idleCount}个，
  32. * 或者池对象的总数(空闲、检出、被创建)达到{@link #getMaxTotal()}。
  33. * 如果{@code always}是false，则不会创建实例，除非线程在等待对象池中的实例检出。
  34. *
  35. * @param idleCount the number of idle instances desired (期望的空闲实例数量)
  36. * @param always true means create instances even if the pool has no threads waiting
  37. *          (true意味着即使对象池没有线程等待，也会创建实例)
  38. * @throws Exception if the factory's makeObject throws
  39. */
  40. private void ensureIdle(int idleCount, boolean always) throws Exception {
  41. if (idleCount < 1 || this.isClosed() || (!always && !idleObjects.hasTakeWaiters())) {
  42. return;
  43. }
  44. while (idleObjects.size() < idleCount) {
  45. PooledObject<T> p = this.create();
  46. if (p == null) {
  47. // Can't create objects (不能创建对象), no reason to think another call to
  48. // create will work. Give up.
  49. break;
  50. }
  51. // "新的池对象"可以立刻被使用
  52. if (this.getLifo()) { // LIFO(后进先出)
  53. idleObjects.addFirst(p);
  54. } else { // FIFO(先进先出)
  55. idleObjects.addLast(p);
  56. }
  57. }
  58. }
  59. /**
  60. * 尝试着创建一个新的包装的池对象。
  61. *
  62. * @return The new wrapped pooled object
  63. *
  64. * @throws Exception if the object factory's {@code makeObject} fails
  65. */
  66. private PooledObject<T> create() throws Exception {
  67. // 1. 对象池是否被耗尽判断
  68. int localMaxTotal = getMaxTotal();
  69. long newCreateCount = createCount.incrementAndGet();
  70. if (localMaxTotal > -1 && newCreateCount > localMaxTotal ||
  71. newCreateCount > Integer.MAX_VALUE) {
  72. createCount.decrementAndGet();
  73. return null; // 没有池对象可创建
  74. }
  75. final PooledObject<T> p;
  76. try {
  77. // 2. 使用PooledObjectFactory.makeObject()来制造一个新的池对象
  78. p = factory.makeObject();
  79. } catch (Exception e) {
  80. createCount.decrementAndGet();
  81. throw e;
  82. }
  83. AbandonedConfig ac = this.abandonedConfig;
  84. if (ac != null && ac.getLogAbandoned()) {
  85. p.setLogAbandoned(true);
  86. }
  87. createdCount.incrementAndGet();
  88. // 3. 将新创建的池对象追加到"池的所有对象映射表"中
  89. allObjects.put(p.getObject(), p);
  90. return p;
  91. }

3.1 "驱逐回收策略"实现

EvictionConfig："驱逐回收策略"配置信息

  1. /**
  2. * 此类用于将对象池的配置信息传递给"驱逐回收策略({@link EvictionPolicy})"实例。
  3. * <p>
  4. * <font color="red">此类是不可变的，且是线程安全的。</font>
  5. *
  6. * @since 2.0
  7. */
  8. public class EvictionConfig {
  9. // final 字段修饰保证其不可变性
  10. /** 池对象的最大空闲驱逐时间（当池对象的空闲时间超过该值时，立马被强制驱逐掉） */
  11. private final long idleEvictTime;
  12. /** 池对象的最小空闲驱逐时间（当池对象的空闲时间超过该值时，被纳入驱逐对象列表里） */
  13. private final long idleSoftEvictTime;
  14. /** 对象池的最小空闲池对象数量 */
  15. private final int minIdle;
  16. /**
  17. * 创建一个新的"驱逐回收策略"配置实例。
  18. * <p>
  19. * <font color="red">实例是不可变的。</font>
  20. *
  21. * @param poolIdleEvictTime Expected to be provided by (池对象的最大空闲驱逐时间(ms))
  22. *        {@link BaseGenericObjectPool#getMinEvictableIdleTimeMillis()}
  23. * @param poolIdleSoftEvictTime Expected to be provided by (池对象的最小空闲驱逐时间(ms))
  24. *        {@link BaseGenericObjectPool#getSoftMinEvictableIdleTimeMillis()}
  25. * @param minIdle Expected to be provided by (对象池的最小空闲池对象数量)
  26. *        {@link GenericObjectPool#getMinIdle()} or
  27. *        {@link GenericKeyedObjectPool#getMinIdlePerKey()}
  28. */
  29. public EvictionConfig(long poolIdleEvictTime, long poolIdleSoftEvictTime,
  30. int minIdle) {
  31. if (poolIdleEvictTime > 0) {
  32. idleEvictTime = poolIdleEvictTime;
  33. } else {
  34. idleEvictTime = Long.MAX_VALUE;
  35. }
  36. if (poolIdleSoftEvictTime > 0) {
  37. idleSoftEvictTime = poolIdleSoftEvictTime;
  38. } else {
  39. idleSoftEvictTime  = Long.MAX_VALUE;
  40. }
  41. this.minIdle = minIdle;
  42. }
  43. /**
  44. * 获取"池对象的最大空闲驱逐时间(ms)"。
  45. * <p>
  46. * 当池对象的空闲时间超过该值时，立马被强制驱逐掉。
  47. * <p>
  48. * How the evictor behaves based on this value will be determined by the
  49. * configured {@link EvictionPolicy}.
  50. *
  51. * @return The {@code idleEvictTime} in milliseconds
  52. */
  53. public long getIdleEvictTime() {
  54. return idleEvictTime;
  55. }
  56. /**
  57. * 获取"池对象的最小空闲驱逐时间(ms)"。
  58. * <p>
  59. * 当池对象的空闲时间超过该值时，被纳入驱逐对象列表里。
  60. * <p>
  61. * How the evictor behaves based on this value will be determined by the
  62. * configured {@link EvictionPolicy}.
  63. *
  64. * @return The (@code idleSoftEvictTime} in milliseconds
  65. */
  66. public long getIdleSoftEvictTime() {
  67. return idleSoftEvictTime;
  68. }
  69. /**
  70. * 获取"对象池的最小空闲池对象数量"。
  71. * <p>
  72. * How the evictor behaves based on this value will be determined by the
  73. * configured {@link EvictionPolicy}.
  74. *
  75. * @return The {@code minIdle}
  76. */
  77. public int getMinIdle() {
  78. return minIdle;
  79. }
  80. }

EvictionPolicy<T>："驱逐回收策略"声明

  1. /**
  2. * 为了提供对象池的一个自定义"驱逐回收策略"，
  3. * 使用者必须提供该接口的一个实现(如{@link DefaultEvictionPolicy})。
  4. *
  5. * @param <T> the type of objects in the pool (对象池中对象的类型)
  6. *
  7. * @since 2.0
  8. */
  9. public interface EvictionPolicy<T> {
  10. /**
  11. * 一个对象池中的空闲对象是否应该被驱逐，调用此方法来测试。（驱逐行为声明）
  12. *
  13. * @param config    The pool configuration settings related to eviction (与驱逐相关的对象池配置设置)
  14. * @param underTest The pooled object being tested for eviction (正在被驱逐测试的池对象)
  15. * @param idleCount The current number of idle objects in the pool including
  16. *                      the object under test (当前对象池中的空闲对象数，包括测试中的对象)
  17. * @return <code>true</code> if the object should be evicted, otherwise
  18. *             <code>false</code> (如果池对象应该被驱逐掉，就返回{@code true}；否则，返回{@code false}。)
  19. */
  20. boolean evict(EvictionConfig config, PooledObject<T> underTest,
  21. int idleCount);
  22. }

DefaultEvictionPolicy<T>：提供用在对象池的"驱逐回收策略"的默认实现，继承自EvictionPolicy<T>

  1. /**
  2. * 提供用在对象池的"驱逐回收策略"的默认实现，继承自{@link EvictionPolicy}。
  3. * <p>
  4. * 如果满足以下条件，对象将被驱逐：
  5. * <ul>
  6. * <li>池对象的空闲时间超过{@link GenericObjectPool#getMinEvictableIdleTimeMillis()}
  7. * <li>对象池中的空闲对象数超过{@link GenericObjectPool#getMinIdle()}，且池对象的空闲时间超过{@link GenericObjectPool#getSoftMinEvictableIdleTimeMillis()}
  8. * </ul>
  9. * <font color="red">此类是不可变的，且是线程安全的。</font>
  10. *
  11. * @param <T> the type of objects in the pool (对象池中对象的类型)
  12. *
  13. * @since 2.0
  14. */
  15. public class DefaultEvictionPolicy<T> implements EvictionPolicy<T> {
  16. /**
  17. * 如果对象池中的空闲对象是否应该被驱逐，调用此方法来测试。（驱逐行为实现）
  18. */
  19. @Override
  20. public boolean evict(EvictionConfig config, PooledObject<T> underTest,
  21. int idleCount) {
  22. if ((idleCount > config.getMinIdle() &&
  23. underTest.getIdleTimeMillis() > config.getIdleSoftEvictTime())
  24. || underTest.getIdleTimeMillis() > config.getIdleEvictTime()) {
  25. return true;
  26. }
  27. return false;
  28. }
  29. }

其他相关实现

  1. // --- internal attributes (内部属性) -------------------------------------------------
  2. /** 对象池中的所有池对象映射表 */
  3. private final ConcurrentMap<T, PooledObject<T>> allObjects =
  4. new ConcurrentHashMap<T, PooledObject<T>>();
  5. /** 池的空闲池对象列表 */
  6. private final LinkedBlockingDeque<PooledObject<T>> idleObjects =
  7. new LinkedBlockingDeque<PooledObject<T>>();
  8. /** 池对象工厂 */
  9. private final PooledObjectFactory<T> factory;
  10. /**
  11. * 计算空闲对象驱逐者一轮测试的对象数量。
  12. *
  13. * @return The number of objects to test for validity (要测试其有效性的对象数量)
  14. */
  15. private int getNumTests() {
  16. int numTestsPerEvictionRun = this.getNumTestsPerEvictionRun();
  17. if (numTestsPerEvictionRun >= 0) {
  18. return Math.min(numTestsPerEvictionRun, idleObjects.size());
  19. } else {
  20. return (int) (Math.ceil(idleObjects.size() /
  21. Math.abs((double) numTestsPerEvictionRun)));
  22. }
  23. }
  24. /**
  25. * 销毁一个包装的"池对象"。
  26. *
  27. * @param toDestory The wrapped pooled object to destroy
  28. *
  29. * @throws Exception If the factory fails to destroy the pooled object
  30. *                   cleanly
  31. */
  32. private void destroy(PooledObject<T> toDestory) throws Exception {
  33. // 1. 设置这个"池对象"的状态为"无效(INVALID)"
  34. toDestory.invalidate();
  35. // 2. 将这个"池对象"从"空闲对象列表"和"所有对象列表"中移除掉
  36. idleObjects.remove(toDestory);
  37. allObjects.remove(toDestory.getObject());
  38. try {
  39. // 3. 使用PooledObjectFactory.destroyObject(PooledObject<T> p)来销毁这个不再需要的池对象
  40. factory.destroyObject(toDestory);
  41. } finally {
  42. destroyedCount.incrementAndGet();
  43. createCount.decrementAndGet();
  44. }
  45. }
  46. /**
  47. * 恢复被废弃的对象，它已被检测出超过{@code AbandonedConfig#getRemoveAbandonedTimeout()
  48. * removeAbandonedTimeout}未被使用。
  49. * <p>
  50. * <font color="red">注意：需要考虑性能损耗，因为它会对对象池中的所有池对象进行检测！</font>
  51. *
  52. * @param ac The configuration to use to identify abandoned objects
  53. */
  54. private void removeAbandoned(AbandonedConfig ac) {
  55. // 1. Generate a list of abandoned objects to remove (生成一个要被删除的被废弃的对象列表)
  56. final long now = System.currentTimeMillis();
  57. final long timeout =
  58. now - (ac.getRemoveAbandonedTimeout() * 1000L);
  59. List<PooledObject<T>> remove = new ArrayList<PooledObject<T>>();
  60. Iterator<PooledObject<T>> it = allObjects.values().iterator();
  61. while (it.hasNext()) {
  62. PooledObject<T> pooledObject = it.next();
  63. synchronized (pooledObject) {
  64. // 从"所有池对象"中挑选出状态为"使用中"的池对象，且空闲时间已超过了"对象的移除超时时间"
  65. if (pooledObject.getState() == PooledObjectState.ALLOCATED &&
  66. pooledObject.getLastUsedTime() <= timeout) {
  67. // 标记池对象为"被废弃"状态，并添加到删除列表中
  68. pooledObject.markAbandoned();
  69. remove.add(pooledObject);
  70. }
  71. }
  72. }
  73. // 2. Now remove the abandoned objects (移除所有被废弃的对象)
  74. Iterator<PooledObject<T>> itr = remove.iterator();
  75. while (itr.hasNext()) {
  76. PooledObject<T> pooledObject = itr.next();
  77. if (ac.getLogAbandoned()) {
  78. pooledObject.printStackTrace(ac.getLogWriter());
  79. }
  80. try {
  81. this.invalidateObject(pooledObject.getObject());
  82. } catch (Exception e) {
  83. e.printStackTrace();
  84. }
  85. }
  86. }

综上所述，真正的"空闲对象的驱逐检测操作"在 GenericObjectPool.evict() 中实现，其被包装在"驱逐者定时器任务(Evictor)"中，并由"驱逐定时器(EvictionTimer)"定时调度，而启动"驱逐者线程"则由 BaseGenericObjectPool.startEvictor(long delay) 实现。

http://bert82503.iteye.com/blog/2171595

上一篇：struts2框架


下一篇：Spring Boot笔记(四） springboot 集成 @Scheduled 定时任务