美文网首页HiKariCPHikariCP 源码分析系列
HikariCP源码分析之ConcurrentBag一

HikariCP源码分析之ConcurrentBag一

作者: 枫山别院 | 来源:发表于2020-08-05 13:47 被阅读0次

    欢迎访问我的博客,同步更新: 枫山别院

    源代码版本2.4.5-SNAPSHOT

    大家好,今天我们一起分析下 HikariCP 的核心ConcurrentBag,它是管理连接池的最重要的核心类。从它的名字大家可以看得出来,它是一个并发管理类,性能非常好,这是它性能甩其他连接池十条街的秘密所在。

    代码概览

    我们先看一下代码,注意这不是全部的代码,省略了不太重要的部分。大家可以看到我加了非常详细的注释,对详解不太感兴趣的朋友可以直接读一下代码即可,不过这部分历时好几个夜晚我才写完,大家可以稍稍捧个场:

    //可用连接同步器, 用于线程间空闲连接数的通知, synchronizer.currentSequence()方法可以获取当前数量
    //其实就是一个计数器, 连接池中创建了一个连接或者还回了一个连接就 + 1, 但是连接池的连接被借走, 是不会 -1 的, 只加不减
    //用于在线程从连接池中获取连接时, 查询是否有空闲连接添加到连接池, 详见borrow方法
    private final QueuedSequenceSynchronizer synchronizer;
    //sharedList保存了所有的连接
    private final CopyOnWriteArrayList<T> sharedList;
    //threadList可能会保存sharedList中连接的引用
    private final ThreadLocal<List<Object>> threadList;
    //对HikariPool的引用, 用于请求创建新连接
    private final IBagStateListener listener;
    //当前等待获取连接的线程数
    private final AtomicInteger waiters;
    //标记连接池是否关闭的状态
    private volatile boolean closed;
    
    
    /**
     * 该方法会从连接池中获取连接, 如果没有连接可用, 会一直等待timeout超时
     *
     * @param timeout  超时时间
     * @param timeUnit 时间单位
     * @return a borrowed instance from the bag or null if a timeout occurs
     * @throws InterruptedException if interrupted while waiting
     */
    public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException {
       //①
       //先尝试从ThreadLocal中获取
       List<Object> list = threadList.get();
       if (weakThreadLocals && list == null) {
          //如果ThreadLocal是 null, 就初始化, 防止后面 npe
          list = new ArrayList<>(16);
          threadList.set(list);
       }
       //②
       //如果ThreadLocal中有连接的话, 就遍历, 尝试获取
       //从后往前反向遍历是有好处的, 因为最后一次使用的连接, 空闲的可能性比较大, 之前的连接可能会被其他线程偷窃走了
       for (int i = list.size() - 1; i >= 0; i--) {
          final Object entry = list.remove(i);
          @SuppressWarnings("unchecked") final T bagEntry = weakThreadLocals ? ((WeakReference<T>) entry).get() : (T) entry;
          if (bagEntry != null && bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
             return bagEntry;
          }
       }
       //③
       //如果没有从ThreadLocal中获取到连接, 那么就sharedList连接池中遍历, 获取连接, timeout时间后超时
       //因为ThreadLocal中保存的连接是当前线程使用过的, 才会在ThreadLocal中保留引用, 连接池中可能还有其他空闲的连接, 所以要遍历连接池
       //看一下requite(final T bagEntry)方法的实现, 还回去的连接放到了ThreadLocal中
       timeout = timeUnit.toNanos(timeout);
       Future<Boolean> addItemFuture = null;
       //记录从连接池获取连接的开始时间, 后面用
       final long startScan = System.nanoTime();
       final long originTimeout = timeout;
       long startSeq;
       //将等待连接的线程计数器加 1
       waiters.incrementAndGet();
       try {
          do {
             // scan the shared list
             do {
                //④
                //当前连接池中的连接数, 在连接池中添加新连接的时候, 该值会增加
                startSeq = synchronizer.currentSequence();
                for (T bagEntry : sharedList) {
                   if (bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
                      // if we might have stolen another thread's new connection, restart the add...
                      //⑤
                      //如果waiters大于 1, 说明除了当前线程之外, 还有其他线程在等待空闲连接
                      //这里, 当前线程的addItemFuture是 null, 说明自己没有请求创建新连接, 但是拿到了连接, 这就说明是拿到了其他线程请求创建的连接, 这就是所谓的偷窃了其他线程的连接, 然后当前线程请求创建一个新连接, 补偿给其他线程
                      if (waiters.get() > 1 && addItemFuture == null) {
                         //提交一个异步添加新连接的任务
                         listener.addBagItem();
                      }
                      return bagEntry;
                   }
                }
             } while (startSeq < synchronizer.currentSequence()); //如果连接池中的空闲连接数量比循环之前多了, 说明有新连接加入, 继续循环获取
             //⑥
             //循环完一遍连接池(也可能循环多次, 如果正好在第一次循环完连接池后有新连接加入, 那么会继续循环), 还是没有能拿到空闲连接, 就请求创建新的连接
             if (addItemFuture == null || addItemFuture.isDone()) {
                addItemFuture = listener.addBagItem();
             }
             //计算 剩余的超时时间 = 用户设置的connectionTimeout - (系统当前时间 - 开始获取连接的时间_代码①处 即从连接池中获取连接一共使用的时间)
             timeout = originTimeout - (System.nanoTime() - startScan);
          } while (timeout > 10_000L && synchronizer.waitUntilSequenceExceeded(startSeq, timeout)); //③
          //⑦
          //这里的循环条件比较复杂
          //1. 如果剩余的超时时间, 大于10_000纳秒
          //2. startSeq的数量, 即空闲连接数超过循环之前的数量
          //3. 没有超过超时时间timeout
          //满足以上 3 个条件才会继续循环, 否则阻塞线程, 直到满足以上条件
          //如果一直等到timeout超时时间用完都没有满足条件, 结束阻塞, 往下走
          //有可能会动态改变的条件, 只有startSeq数量改变, 是②处添加的创建连接请求
       } finally {
          waiters.decrementAndGet();
       }
    
       return null;
    }
    
    /**
     * 该方法将借出去的连接还回到连接池中
     * 不通过该方法还回的连接会造成内存泄露
     *
     * @param bagEntry the value to return to the bag
     * @throws NullPointerException  if value is null
     * @throws IllegalStateException if the requited value was not borrowed from the bag
     */
    public void requite(final T bagEntry) {
       //⑧
       //lazySet方法不能保证连接会立刻被设置成可用状态, 这是个延迟方法
       //这是一种优化, 如果要立即生效的话, 可能会需要使用volatile等, 让其他线程立即发现, 这会降低性能, 使用lazySet浪费不了多少时间, 但是不会浪费性能
       bagEntry.lazySet(STATE_NOT_IN_USE);
    
       //⑨
       //将连接放回到threadLocal中
       final List<Object> threadLocalList = threadList.get();
       if (threadLocalList != null) {
          threadLocalList.add(weakThreadLocals ? new WeakReference<>(bagEntry) : bagEntry);
       }
       //通知等待线程, 有可用连接
       synchronizer.signal();
    }
    
    /**
     * 在连接池中添加一个连接
     * 新连接都是添加到sharedList中, threadList是sharedList中的部分连接的引用
     *
     * @param bagEntry an object to add to the bag
     */
    public void add(final T bagEntry) {
       if (closed) {
          LOGGER.info("ConcurrentBag has been closed, ignoring add()");
          throw new IllegalStateException("ConcurrentBag has been closed, ignoring add()");
       }
       //⑩
       sharedList.add(bagEntry);
       synchronizer.signal();
    }
    
    /**
     * 从连接池中移除一个连接.
     * 这个方法只能用于从<code>borrow(long, TimeUnit)</code> 或者 <code>reserve(T)</code>方法中获取到的连接
     * 也就是说, 这个方法只能移除处于使用中和保留状态的连接
     *
     * @param bagEntry the value to remove
     * @return true if the entry was removed, false otherwise
     * @throws IllegalStateException if an attempt is made to remove an object
     *                               from the bag that was not borrowed or reserved first
     */
    public boolean remove(final T bagEntry) {
       //⑪
       //尝试标记移除使用中和保留状态的连接, 如果标记失败, 就是空闲的连接, 直接返回 false
       //也就是检查连接的状态, 不能移除空闲的连接或者已经标记移除的连接
       if (!bagEntry.compareAndSet(STATE_IN_USE, STATE_REMOVED) && !bagEntry.compareAndSet(STATE_RESERVED, STATE_REMOVED) && !closed) {
          LOGGER.warn("Attempt to remove an object from the bag that was not borrowed or reserved: {}", bagEntry);
          return false;
       }
       //如果上面标记成功了, 那么从连接池中移除这个连接
       final boolean removed = sharedList.remove(bagEntry);
       if (!removed && !closed) {
          LOGGER.warn("Attempt to remove an object from the bag that does not exist: {}", bagEntry);
       }
    
       // synchronizer.signal();
       return removed;
    }
    

    上面的代码是ConcurrentBag中的成员变量和最重要的四个方法,ConcurrentBag中的属性我们穿插在代码中解释。

    borrow方法

    borrow方法应该是整个 HikariCP 中最最核心的方法,它是我们从连接池中获取连接的时候最终会调用到的方法,一切秘密都在这里了。我们分析下:

    ①ThreadLocal

    //①
    //先尝试从ThreadLocal中获取
    List<Object> list = threadList.get();
    if (weakThreadLocals && list == null) {
      //如果ThreadLocal是 null, 就初始化, 防止后面 npe
      list = new ArrayList<>(16);
      threadList.set(list);
    }
    

    threadList是在ConcurrentBag上的成员变量,它的定义是private final ThreadLocal<List> threadList;,可见它是一个ThreadLocal,也就是每个线程都有独享的一个 List,它是用于保存当前线程用过的连接。注意这里我说的是"用过",不是所有的连接。因为还有一个成员变量private final CopyOnWriteArrayList<T> sharedList;,它是真正的保存所有的连接的地方,它是一个CopyOnWriteArrayList,在写入的时候会先复制一个 sharedList2,然后修改这个新的 sharedList2,最后将变量地址指向新的sharedList2,不是直接写入当前的sharedList,典型的空间换时间的一个做法,可以避免写入前要锁住sharedList,从而导致降低性能。

    HikariCP 使用过的连接,在还回连接池的时候,是直接放在了ThreadLocal中。说到这里,可能会有同学问了:sharedList中保存了所有的连接,当用户借走了一个连接,不是应该把这个连接从sharedList中移除,然后还回来的时候再把连接加入到sharedList中?为什么还回去的时候,没有放到sharedList中呢?

    首先,明确一点,HikariCP 不是这样做的。为什么呢?如果用户借用连接的时候,你从sharedList中移除了,那么相当于这个连接脱离了 HikariCP 的管理,后面 HikariCP 还怎么管理这个连接呢?比如这个连接的生命周期到时间了,连接都让用户拐跑了,我还怎么关闭这个连接呢?所以,所有的连接都不能脱离掌控,一个都不能少。其实,我们在sharedList中保存的仅仅是数据库连接的引用,这些连接是所有的线程都可见的,各个线程也可以随意保存连接的引用,只是要使用的时候必须要走borrow方法,按流程来。

    为什么要放到线程的threadList中?

    因为下次获取的时候比较方便,也许会提高性能。每个线程都优先从自己的本地线程中拿,竞争的可能性大大降低啊,也许这个连接刚刚用完到再次获取的时间极短,这个连接很可能还空闲着。只有在本地线程中的连接都不能使用的时候,才去sharedList这个 HikariCP的总仓库里获取。

    举一个生活例子:假如你是一个连锁店老板,提供汽车出租服务,有一个总仓库,所有的连锁店都从这里提车出租给用户。刚开始,你是每租一辆车都去仓库直接提货,用户还车的时候,你直接送到仓库。过了一段时间,你觉得这样不行啊,太浪费时间了,而且所有的连锁店都这样,各个店的老板都去提车,太忙了,还得排队。要不用户还回来的车先放店里吧,这样下次有用户租车就不用去仓库了,直接给他,方便很多,店里没车了再去总仓提车。其他连锁店都开始这么搞,大家都先用店里的车不够再去总仓。生意火爆,有一天店里没车了,你去仓库提车,仓库管理员说:仓库也没车了,天通苑的连锁店里有闲着的,你去那里提吧,于是你把天通苑连锁店的车借走了。所以各个连锁店之间也有相互借车。

    例子可能不太恰当,一时也想不到同样道理的生活例子,但是就这个意思。HikariCP 也是这样,用户使用的连接,还回连接池的时候,直接放到线程的本地threadList中,如果用户又要借用连接,先看本地有没有,优先使用本地连接,只有本地没有或者都不可用的时候,再去 HikariCP 的连接池里获取。但是跟借车不同,因为我们本地是保存的sharedList中连接的引用,虽然你还有这个连接的引用,但是很可能它已经被其他线程从sharedList借走了,这就是HikariCP所谓的线程间的连接窃取。所以线程在本地的threadList就算拿到了连接,也必须检查下状态,是不是可用的。

    说到这里,还没有解析代码,扯远了。①出代码就是先从本地的threadList里取出连接的 List,然后检查下List 是否为空,是空的直接初始化一个 List,因为下面要用到,防止抛空指针了。大家可以看到判空的时候,还有一个条件是weakThreadLocals,这个标识是表示threadList是否是弱引用。如果是弱引用,那么很可能 GC 的时候会被回收掉,所以变成 null 了,但是如果不是弱引用的话,那么它是在初始化ConcurrentBag的时候,就是一个FastList了,不用担心是 null。那么什么情况下threadList会是弱引用呢?当 HikariCP 运行在容器中时,会使用弱引用,因为在容器重新部署的时候,可能会导致发成内存泄露,具体大家可以看下#39 的 issue。

    相关文章

      网友评论

        本文标题:HikariCP源码分析之ConcurrentBag一

        本文链接:https://www.haomeiwen.com/subject/kundrktx.html