数据库连接池之Hikari源码解析-飞外

Hikari连接池目前公认是性能最高的数据库连接池，同时也是SpringBoot2.0以后默认使用的数据库连接池。

由于Springboot2.0默认就是使用的Hikari连接池，所以无需额外添加Hikari相关的maven依赖。只需要在application.yml添加对应的配置即可，如下：

spring.datasource.driver-class-name=com.mysql.jdbc.Driverspring.datasource.url=jdbc:mysql://127.0.0.1:3306/test?useUnicode=true characterEncoding=UTF-8 autoReconnect=true useSSL=false zeroDateTimeBehavior=convertToNullspring.datasource.username=adminspring.datasource.password=adminspring.datasource.type=com.zaxxer.hikari.HikariDataSourcespring.datasource.hikari.minimum-idle=5spring.datasource.hikari.maximum-pool-size=15spring.datasource.hikari.auto-commit=truespring.datasource.hikari.idle-timeout=30000spring.datasource.hikari.pool-name=DatebookHikariCPspring.datasource.hikari.max-lifetime=1800000spring.datasource.hikari.connection-timeout=30000spring.datasource.hikari.connection-test-query=SELECT 1

1、Hikari中的核心类为HikariDataSource，表示Hikari连接池中的数据源,实现了DataSource接口的getConnection方法，getConnection方法源码如下：

 1 /** 连接池对象 2 * fastPathPool 会在初始化时创建 3 * pool 是在获取连接数创建 4 * volatile修饰pool导致每次读pool都要从主存加载，每次写也要写回主存，性能不如没volatile修饰的fastPathPool 5 * */ 6 private final HikariPool fastPathPool; 7 private volatile HikariPool pool; 9 /** 获取连接*/10 public Connection getConnection() throws SQLException12 if (isClosed()) {13 throw new SQLException("HikariDataSource " + this + " has been closed.");15 /** 如果fastPathPool存在则直接获取连接 */16 if (fastPathPool != null) {17 return fastPathPool.getConnection();19 /** 如果没有fastPathPool 则创建HikariPool对象 */20 HikariPool result = pool;21 if (result == null) {22 synchronized (this) {23 result = pool;24 if (result == null) {25 validate();26 LOGGER.info("{} - Starting...", getPoolName());27 try {28 /** 初始化创建HikariPool对象*/29 pool = result = new HikariPool(this);30 this.seal();32 catch (PoolInitializationException pie) {33 //38 /** 调用pool的getConnection()方法获取连接*/39 return result.getConnection();40 }

getConnection方法逻辑不多，主要是调用了HikariPool的getConnection()方法，而HikariDataSource中有两个HikariPool对象,一个是fastPathPool是在HikariPool有参构造函数中创建, 如果没有创建fastPathPool,那么就会在getConnection方法时创建pool对象。

很显然pool对象是由volatile关键字修饰的，而fastPathPool是final类型的，所以fastPathPool的效率会比pool要高，所以推荐使用HikariDataSource有参构造函数进行初始化。

2、由上可知获取连接的逻辑是在HikariPool的getConnection方法中,继续分析HikariPool的getConnection方法，源码如下：

 1 /** 获取连接*/ 2 public Connection getConnection(final long hardTimeout) throws SQLException 4 /** 获取锁*/ 5 suspendResumeLock.acquire(); 6 final long startTime = currentTime(); 8 try { 9 long timeout = hardTimeout;10 do {11 /** 从ConcurrentBag中借出一个PoolEntry对象 */12 PoolEntry poolEntry = connectionBag.borrow(timeout, MILLISECONDS);13 if (poolEntry == null) {14 break; // We timed out... break and throw exception17 final long now = currentTime();18 /** 判断连接是否被标记为抛弃 或者 空闲时间过长, 是的话就关闭连接*/19 if (poolEntry.isMarkedEvicted() || (elapsedMillis(poolEntry.lastAccessed, now) aliveBypassWindowMs !isConnectionAlive(poolEntry.connection))) {20 closeConnection(poolEntry, poolEntry.isMarkedEvicted() ? EVICTED_CONNECTION_MESSAGE : DEAD_CONNECTION_MESSAGE);21 timeout = hardTimeout - elapsedMillis(startTime);23 else {24 metricsTracker.recordBorrowStats(poolEntry, startTime);25 /** 通过Javassist创建代理连接*/26 return poolEntry.createProxyConnection(leakTaskFactory.schedule(poolEntry), now);28 } while (timeout 0L);29 metricsTracker.recordBorrowTimeoutStats(startTime);30 throw createTimeoutException(startTime);32 catch (InterruptedException e) {33 Thread.currentThread().interrupt();34 throw new SQLException(poolName + " - Interrupted during connection acquisition", e);36 finally {37 /** 释放锁*/38 suspendResumeLock.release();40 }

核心步骤只有两步，一个是调用ConcurrentBag的borrow方法借用一个PoolEntry对象，第二步调用调用PoolEntry的createProxyConnection方法动态生成代理connection对象。

这里涉及到了两个核心的类，分别是ConcurrentBag和PoolEntry

3、PoolEntry

PoolEntry顾名思义是连接池的节点，实际也可以看作是一个Connection对象的封装，连接池中存储的连接就是以PoolEntry的方式进行存储。

PoolEntry内部属性如下：

4、ConcurrentBag

ConcurrentBag直意就是并发包，本质就是连接池的主体，存储连接的封装对象PoolEntry，另外做了并发控制来解决连接池的并发问题。

ConcurrentBag的内部属性如下：

5、从ConcurrentBag借出一个元素

ConcurrentBag实现了borrow方法，意思是从并发集合中借出一个元素，对于连接池而言实际就是从连接池中获取一个连接，源码如下：

 1 /** 借出一个对象 */ 2 public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException 4 /** 1.从ThreadLocal中获取当前线程绑定的对象集合 */ 5 final List Object list = threadList.get(); 6 /** 1.1.如果当前线程变量中存在就直接从list中返回一个*/ 7 for (int i = list.size() - 1; i i--) { 8 final Object entry = list.remove(i); 9 @SuppressWarnings("unchecked")10 final T bagEntry = weakThreadLocals ? ((WeakReference T ) entry).get() : (T) entry;11 if (bagEntry != null bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {12 return bagEntry;16 /** 2.当前等待对象数量自增1 */17 final int waiting = waiters.incrementAndGet();18 try {19 /** 3.遍历当前缓存的sharedList, 如果当前状态为未使用,则通过CAS修改为已使用*/20 for (T bagEntry : sharedList) {21 if (bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {22 /** 4.如果当前等待线程不止1个,则给监听中添加一个任务 */23 if (waiting 1) {24 listener.addBagItem(waiting - 1);26 return bagEntry;30 /** 4.如果当前缓存的sharedList为空或者都在使用中,那么给listener添加一个任务*/31 listener.addBagItem(waiting);33 timeout = timeUnit.toNanos(timeout);34 do {35 final long start = currentTime();36 /** 5.从阻塞队列中等待超时获取元素 */37 final T bagEntry = handoffQueue.poll(timeout, NANOSECONDS);38 /** 6.如果获取元素失败或者获取元素且使用成功则均返回 */39 if (bagEntry == null || bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {40 return bagEntry;43 timeout -= elapsedNanos(start);44 } while (timeout 10_000);46 return null;48 finally {49 /** 6.等待线程数自减1 */50 waiters.decrementAndGet();52 }

从源码中可以发现方法中共有三个地方出现了return bagEntry，所以可以看出ConcurrentBag借出元素的地方是有三个来源的

第一步：从ThreadLocal中获取

每个线程从ConcurrentBag中借出连接时都会创建一个ThreadLocal对象,值为一个List，默认大小为16，每次客户端从ConcurrentBag中获取连接时都会优先从ThreadLocal中尝试获取连接，获取失败才会走下一步获取。

当客户端将连接归还给ConcurrentBag时，首先判断当前是否有其他客户端等待连接，如果有其他客户端等待那么就将连接给其他客户端，如果没有客户端等待那么将连接存入ThreadLocal中，每个ThreadLocal最多会存储50个连接

Tips：使用ThreadLocal可能会存在内存泄露的风险，所以ConcurrentBag内部有一个属性为booleal weakThreadLocals，当值为true时则ThreadLocal中的引用均是弱引用，在内存不足时GC的时候会被回收，避免了出现内存泄露的问题。

第二步：从sharedList中获取

从ThreadLocal中获取连接失败之后，会再次尝试从sharedList中获取，sharedList集合存在初始化的PoolEntry。在ConcurrentBag初始化的，会初始化指定数量的PoolEntry对象存入sharedList，源码如下：

ConcurrentBag构造函数如下：

 1 /** ConcurrentHand 2 * IBagStateListener bag状态监听器,HikariPool实现了IBagStateListener接口 3 * 所以构造器传入的listener实际就是HikariPool对象 4 * */ 5 public ConcurrentBag(final IBagStateListener listener) 7 this.listener = listener; 8 //是否使用弱引用 9 this.weakThreadLocals = useWeakThreadLocals();10 //初始化阻塞队列11 this.handoffQueue = new SynchronousQueue (true);12 //初始化等待连接数13 this.waiters = new AtomicInteger();14 //初始化sharedList15 this.sharedList = new CopyOnWriteArrayList ();16 if (weakThreadLocals) {17 this.threadList = ThreadLocal.withInitial(() - new ArrayList (16));19 else {20 this.threadList = ThreadLocal.withInitial(() - new FastList (IConcurrentBagEntry.class, 16));22 }

HikariPool内部属性包含了ConcurrentBag对象,在HikariPool初始化时会创建ConcurrentBag对象,所以ConcurrentBag的构造函数是在HikariPool初始化时调用，HikariPool构造函数如下：

 1 public HikariPool(final HikariConfig config) 3 super(config); 5 //初始化ConcurrentBag对象 6 this.connectionBag = new ConcurrentBag (this); 7 //创建SuspendResumeLock对象 8 this.suspendResumeLock = config.isAllowPoolSuspension() ? new SuspendResumeLock() : SuspendResumeLock.FAUX_LOCK; 9 /** 初始化线程池,houseKeeping可以理解为保持空间充足的意思,空间也就是连接池,该线程池的作用就是保持连接池中合适的连接数的作用 */10 this.houseKeepingExecutorService = initializeHouseKeepingExecutorService();12 /** 设置属性*/13 checkFailFast();15 if (config.getMetricsTrackerFactory() != null) {16 setMetricsTrackerFactory(config.getMetricsTrackerFactory());18 else {19 setMetricRegistry(config.getMetricRegistry());22 setHealthCheckRegistry(config.getHealthCheckRegistry());24 handleMBeans(this, true);26 ThreadFactory threadFactory = config.getThreadFactory();27 /** 根据配置的最大连接数,创建链表类型阻塞队列 */28 LinkedBlockingQueue Runnable addQueue = new LinkedBlockingQueue (config.getMaximumPoolSize());29 this.addConnectionQueue = unmodifiableCollection(addQueue);30 /** 初始化创建连接线程池*/31 this.addConnectionExecutor = createThreadPoolExecutor(addQueue, poolName + " connection adder", threadFactory, new ThreadPoolExecutor.DiscardPolicy());32 /** 初始化关闭连接线程池*/33 this.closeConnectionExecutor = createThreadPoolExecutor(config.getMaximumPoolSize(), poolName + " connection closer", threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());35 this.leakTaskFactory = new ProxyLeakTaskFactory(config.getLeakDetectionThreshold(), houseKeepingExecutorService);36 /** 创建保持连接池连接数量的任务*/37 this.houseKeeperTask = houseKeepingExecutorService.scheduleWithFixedDelay(new HouseKeeper(), 100L, housekeepingPeriodMs, MILLISECONDS);39 if (Boolean.getBoolean("com.zaxxer.hikari.blockUntilFilled") config.getInitializationFailTimeout() 1) {40 addConnectionExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors());41 addConnectionExecutor.setMaximumPoolSize(Runtime.getRuntime().availableProcessors());43 final long startTime = currentTime();44 while (elapsedMillis(startTime) config.getInitializationFailTimeout() getTotalConnections() config.getMinimumIdle()) {45 quietlySleep(MILLISECONDS.toMillis(100));48 addConnectionExecutor.setCorePoolSize(1);49 addConnectionExecutor.setMaximumPoolSize(1);51 }

这里有一个定时任务houseKeeperTask，该定时任务的作用是定时检测连接池中连接的数量，执行的内容就是HouseKeep的run方法，逻辑如下：

 1 private final class HouseKeeper implements Runnable 3 private volatile long previous = plusMillis(currentTime(), -housekeepingPeriodMs); 5 @Override 6 public void run() 8 try { 9 /** 读取连接池配置 */10 connectionTimeout = config.getConnectionTimeout();11 validationTimeout = config.getValidationTimeout();12 leakTaskFactory.updateLeakDetectionThreshold(config.getLeakDetectionThreshold());13 catalog = (config.getCatalog() != null !config.getCatalog().equals(catalog)) ? config.getCatalog() : catalog;15 final long idleTimeout = config.getIdleTimeout();16 final long now = currentTime();18 // Detect retrograde time, allowing +128ms as per NTP spec.19 if (plusMillis(now, 128) plusMillis(previous, housekeepingPeriodMs)) {20 logger.warn("{} - Retrograde clock change detected (housekeeper delta={}), soft-evicting connections from pool.",21 poolName, elapsedDisplayString(previous, now));22 previous = now;23 /** 关闭连接池中需要被丢弃的连接 */24 softEvictConnections();25 return;27 else if (now plusMillis(previous, (3 * housekeepingPeriodMs) / 2)) {28 // No point evicting for forward clock motion, this merely accelerates connection retirement anyway29 logger.warn("{} - Thread starvation or clock leap detected (housekeeper delta={}).", poolName, elapsedDisplayString(previous, now));32 previous = now;34 String afterPrefix = "Pool ";35 if (idleTimeout 0L config.getMinimumIdle() config.getMaximumPoolSize()) {36 logPoolState("Before cleanup ");37 afterPrefix = "After cleanup ";39 /** 获取当前连接池中已经不是使用中的连接集合 */40 final List PoolEntry notInUse = connectionBag.values(STATE_NOT_IN_USE);41 int toRemove = notInUse.size() - config.getMinimumIdle();42 for (PoolEntry entry : notInUse) {43 /** 当前空闲的连接如果超过最大空闲时间idleTimeout则关闭空闲连接 */44 if (toRemove 0 elapsedMillis(entry.lastAccessed, now) idleTimeout connectionBag.reserve(entry)) {45 closeConnection(entry, "(connection has passed idleTimeout)");46 toRemove--;51 logPoolState(afterPrefix);52 /** 填充连接池,保持连接池数量至少保持minimum个连接数量 */53 fillPool(); // Try to maintain minimum connections55 catch (Exception e) {56 logger.error("Unexpected exception in housekeeping task", e);59 }

该定时任务主要是为了维护连接池中连接的数量，首先需要将被标记为需要丢弃的连接进行关闭，然后将空闲超时的连接进行关闭，最后当连接池中的连接少于最小值时就需要对连接池进行补充连接的操作。所以在初始化连接池时，初始化连接的操作就是在fillPool方法中实现的。fillPool方法源码如下：

 1 /** 填充连接池 */ 2 private synchronized void fillPool() 4 /** 5 * 计算需要添加的连接数量 6 * config.getMaximumPoolSize - getTotalConnections() 表示连接池最大值-当前连接的数量=最多还可以创建的连接数 7 * config.getMinimumIdle() - getIdleConnections() 表示连接池最小值 - 当前空闲的连接数= 当前可以连接数 8 * Math.min计算得到最少需要的连接数 - addConnectionQueue.size() = 还需要创建连接的任务数量 9 * */10 final int connectionsToAdd = Math.min(config.getMaximumPoolSize() - getTotalConnections(), config.getMinimumIdle() - getIdleConnections())11 - addConnectionQueue.size();12 for (int i = 0; i connectionsToAdd; i++) {13 /** 向创建连接线程池中提交创建连接的任务 */14 addConnectionExecutor.submit((i connectionsToAdd - 1) ? poolEntryCreator : postFillPoolEntryCreator);16 }

先计算需要创建的连接数量，向创建连接的线程池中提交任务 poolEntryCreator，创建最后一个任务时创建的是postFillPoolEntryCreator, 两者没有本质的区别，只是打印的日志不一样而已.

PoolEntryCreator创建PoolEntry对象的逻辑如下：

 1 /** 创建PoolEntry对象线程 */ 2 private final class PoolEntryCreator implements Callable Boolean { 3 /** 4 * 日志前缀 5 */ 6 private final String loggingPrefix; 8 PoolEntryCreator(String loggingPrefix) { 9 this.loggingPrefix = loggingPrefix;12 @Override13 public Boolean call() {14 long sleepBackoff = 250L;15 /** 1.当前连接池状态正常并且需求创建连接时 */16 while (poolState == POOL_NORMAL shouldCreateAnotherConnection()) {17 /** 2.创建PoolEntry对象 */18 final PoolEntry poolEntry = createPoolEntry();19 if (poolEntry != null) {20 /** 3.将PoolEntry对象添加到ConcurrentBag对象中的sharedList中 */21 connectionBag.add(poolEntry);22 logger.debug("{} - Added connection {}", poolName, poolEntry.connection);23 if (loggingPrefix != null) {24 logPoolState(loggingPrefix);26 return Boolean.TRUE;28 /** 睡眠指定时间*/29 quietlySleep(sleepBackoff);30 sleepBackoff = Math.min(SECONDS.toMillis(10), Math.min(connectionTimeout, (long) (sleepBackoff * 1.5)));32 // Pool is suspended or shutdown or at max size33 return Boolean.FALSE;35 }

首先创建一个新的PoolEntry对象，PoolEntry构造时会创建Connection对象，另外如果连接设置了最大生命周期时长，那么需要给每个PoolEntry添加定时任务，为了防止多个PoolEntry同时创建同时被关闭，所以每个PoolEntry的最大生命周期时间都不一样。当PoolEntry达到最大生命周期后会触发softEvictConnection方法，将PoolEntry标记为需要被丢弃，另外由于抛弃了PoolEntry对象，所以需要重新调用addBagItem方法对PoolEntry对象进行补充。

第三步：通过IBagStateListener创建新的元素

由于第二步可知，IBagStateListener主要有一个addBagItem方法，HikariPool实现了addBagItem方法，方法源码如下：

1 public void addBagItem(final int waiting)3 /** 判断是否需要创建连接 */4 final boolean shouldAdd = waiting - addConnectionQueue.size() // Yes, = is intentional.5 if (shouldAdd) {6 /** 向创建连接线程池中提交创建连接的任务 */7 addConnectionExecutor.submit(poolEntryCreator);9 }

总结：

从ConcurrentBag中获取连接一共分成三步，首先从当前线程的ThreadLocal中获取，如果有直接返回一个连接，如果ThreadLocal中没有则从sharedList中获取，sharedList可以理解为ConcurrentBag缓存的连接池，每当创建了一个PoolEntry对象之后都会添加到sharedList中去，如果sharedList中的连接状态都不是可用状态，此时就需要通过IBagStateListener提交一个创建连接的任务，交给创建连接的线程池去执行，创建新的连接。

新的连接创建成功之后会将PoolEntry对象添加到无容量的阻塞队列handoffQueue中，等待连接的线程不断尝试从handoffQueue队列中获取连接直到成功获取或者超时返回。

当客户端释放连接时会调用collection的close方法，Hikari中的Connection使用的是代理连接ProxyConnection对象，调用close方法时会调用关联的PoolEntry对象的回收方法recycle方法，PoolEntry的recycle方法源码如下：

1 void recycle(final long lastAccessed)3 if (connection != null) {4 this.lastAccessed = lastAccessed;5 /** 调用HikariPool的recycle方法,回收当前PoolEntry对象 */6 hikariPool.recycle(this);8 }

回收连接最终会调用ConcurrentBag的requite方法，方法逻辑不复杂，首先将PoolEntry元素状态设置为未使用，然后判断当前是否存在等待连接的线程，如果存在则将连接加入到无界阻塞队列中去，由等待连接的线程从阻塞队列中去获取；

如果当前没有等待连接的线程，则将连接添加到本地线程变量ThreadLocal中，等待当前线程下次获取连接时直接从ThreadLocal中获取。

1、采用自定义的FastList替代了ArrayList，FastList的get方法去除了范围检查rangeCheck逻辑，并且remove方法是从尾部开始扫描的，而并不是从头部开始扫描的。因为Connection的打开和关闭顺序通常是相反的

2、初始化时创建了两个HikariPool对象，一个采用final类型定义，避免在获取连接时才初始化，因为获取连接时才初始化就需要做同步处理

3、Hikari创建连接是通过javassist动态字节码生成技术创建的，性能更好

4、从连接池中获取连接时对于同一个线程在threadLocal中添加了缓存，同一线程获取连接时没有并发操作

5、Hikari最大的特点是在高并发的情况下尽量的减少锁竞争