数据库连接池之Hikari源码解析（转）

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

 1 /** 连接池对象
 2      * fastPathPool 会在初始化时创建
 3      * pool 是在获取连接数创建
 4      * volatile修饰pool导致每次读pool都要从主存加载，每次写也要写回主存，性能不如没volatile修饰的fastPathPool
 5      * */
 6     private final HikariPool fastPathPool;
 7     private volatile HikariPool pool;
 8 
 9     /** 获取连接*/
10     public Connection getConnection() throws SQLException
11     {
12         if (isClosed()) {
13             throw new SQLException("HikariDataSource " + this + " has been closed.");
14         }
15         /** 如果fastPathPool存在则直接获取连接 */
16         if (fastPathPool != null) {
17             return fastPathPool.getConnection();
18         }
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();
31                     }
32                     catch (PoolInitializationException pie) {
33                         //
34                     }
35                 }
36             }
37         }
38         /** 调用pool的getConnection()方法获取连接*/
39         return result.getConnection();
40     }

 1 public HikariDataSource(HikariConfig configuration)
 2    {
 3       configuration.validate();
 4       configuration.copyStateTo(this);
 5 
 6       LOGGER.info("{} - Starting...", configuration.getPoolName());
 7       pool = fastPathPool = new HikariPool(this);
 8       LOGGER.info("{} - Start completed.", configuration.getPoolName());
 9 
10       this.seal();
11    }

 1 /** 获取连接*/
 2     public Connection getConnection(final long hardTimeout) throws SQLException
 3     {
 4         /** 获取锁*/
 5         suspendResumeLock.acquire();
 6         final long startTime = currentTime();
 7 
 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 exception
15                 }
16 
17                 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);
22                 }
23                 else {
24                     metricsTracker.recordBorrowStats(poolEntry, startTime);
25                     /** 通过Javassist创建代理连接*/
26                     return poolEntry.createProxyConnection(leakTaskFactory.schedule(poolEntry), now);
27                 }
28             } while (timeout > 0L);
29             metricsTracker.recordBorrowTimeoutStats(startTime);
30             throw createTimeoutException(startTime);
31         }
32         catch (InterruptedException e) {
33             Thread.currentThread().interrupt();
34             throw new SQLException(poolName + " - Interrupted during connection acquisition", e);
35         }
36         finally {
37             /** 释放锁*/
38             suspendResumeLock.release();
39         }
40     }

 1 /** 借出一个对象 */
 2     public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException
 3     {
 4         /** 1.从ThreadLocal中获取当前线程绑定的对象集合 */
 5         final List<Object> list = threadList.get();
 6         /** 1.1.如果当前线程变量中存在就直接从list中返回一个*/
 7         for (int i = list.size() - 1; i >= 0; 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;
13             }
14         }
15 
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);
25                     }
26                     return bagEntry;
27                 }
28             }
29 
30             /** 4.如果当前缓存的sharedList为空或者都在使用中,那么给listener添加一个任务*/
31             listener.addBagItem(waiting);
32 
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;
41                 }
42 
43                 timeout -= elapsedNanos(start);
44             } while (timeout > 10_000);
45 
46             return null;
47         }
48         finally {
49             /** 6.等待线程数自减1 */
50             waiters.decrementAndGet();
51         }
52     }

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

 1 public HikariPool(final HikariConfig config)
 2     {
 3         super(config);
 4 
 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();
11 
12         /** 设置属性*/
13         checkFailFast();
14 
15         if (config.getMetricsTrackerFactory() != null) {
16             setMetricsTrackerFactory(config.getMetricsTrackerFactory());
17         }
18         else {
19             setMetricRegistry(config.getMetricRegistry());
20         }
21 
22         setHealthCheckRegistry(config.getHealthCheckRegistry());
23 
24         handleMBeans(this, true);
25 
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());
34 
35         this.leakTaskFactory = new ProxyLeakTaskFactory(config.getLeakDetectionThreshold(), houseKeepingExecutorService);
36         /** 创建保持连接池连接数量的任务*/
37         this.houseKeeperTask = houseKeepingExecutorService.scheduleWithFixedDelay(new HouseKeeper(), 100L, housekeepingPeriodMs, MILLISECONDS);
38 
39         if (Boolean.getBoolean("com.zaxxer.hikari.blockUntilFilled") && config.getInitializationFailTimeout() > 1) {
40             addConnectionExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
41             addConnectionExecutor.setMaximumPoolSize(Runtime.getRuntime().availableProcessors());
42 
43             final long startTime = currentTime();
44             while (elapsedMillis(startTime) < config.getInitializationFailTimeout() && getTotalConnections() < config.getMinimumIdle()) {
45                 quietlySleep(MILLISECONDS.toMillis(100));
46             }
47 
48             addConnectionExecutor.setCorePoolSize(1);
49             addConnectionExecutor.setMaximumPoolSize(1);
50         }
51     }

 1 private final class HouseKeeper implements Runnable
 2     {
 3         private volatile long previous = plusMillis(currentTime(), -housekeepingPeriodMs);
 4 
 5         @Override
 6         public void run()
 7         {
 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;
14 
15                 final long idleTimeout = config.getIdleTimeout();
16                 final long now = currentTime();
17 
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;
26                 }
27                 else if (now > plusMillis(previous, (3 * housekeepingPeriodMs) / 2)) {
28                     // No point evicting for forward clock motion, this merely accelerates connection retirement anyway
29                     logger.warn("{} - Thread starvation or clock leap detected (housekeeper delta={}).", poolName, elapsedDisplayString(previous, now));
30                 }
31 
32                 previous = now;
33 
34                 String afterPrefix = "Pool ";
35                 if (idleTimeout > 0L && config.getMinimumIdle() < config.getMaximumPoolSize()) {
36                     logPoolState("Before cleanup ");
37                     afterPrefix = "After cleanup  ";
38 
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--;
47                         }
48                     }
49                 }
50 
51                 logPoolState(afterPrefix);
52                 /** 填充连接池,保持连接池数量至少保持minimum个连接数量 */
53                 fillPool(); // Try to maintain minimum connections
54             }
55             catch (Exception e) {
56                 logger.error("Unexpected exception in housekeeping task", e);
57             }
58         }
59     }

 1 /** 填充连接池 */
 2     private synchronized void fillPool()
 3     {
 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);
15         }
16     }

 1 /** 创建PoolEntry对象线程 */
 2     private final class PoolEntryCreator implements Callable<Boolean> {
 3         /**
 4          * 日志前缀
 5          */
 6         private final String loggingPrefix;
 7 
 8         PoolEntryCreator(String loggingPrefix) {
 9             this.loggingPrefix = loggingPrefix;
10         }
11 
12         @Override
13         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);
25                     }
26                     return Boolean.TRUE;
27                 }
28                 /** 睡眠指定时间*/
29                 quietlySleep(sleepBackoff);
30                 sleepBackoff = Math.min(SECONDS.toMillis(10), Math.min(connectionTimeout, (long) (sleepBackoff * 1.5)));
31             }
32             // Pool is suspended or shutdown or at max size
33             return Boolean.FALSE;
34         }
35     }

 1 /** 创建PoolEntry对象 */
 2     private PoolEntry createPoolEntry()
 3     {
 4         try {
 5             /** 1.初始化PoolEntry对象,会先创建Connection对象传入PoolEntry的构造函数中 */
 6             final PoolEntry poolEntry = newPoolEntry();
 7             /** 2.获取连接最大生命周期时长 */
 8             final long maxLifetime = config.getMaxLifetime();
 9             if (maxLifetime > 0) {
10                 /** 3.获取一个随机值,防止PoolEntry同时创建同时被销毁,添加随机值错开时间差 */
11                 final long variance = maxLifetime > 10_000 ? ThreadLocalRandom.current().nextLong( maxLifetime / 40 ) : 0;
12                 final long lifetime = maxLifetime - variance;
13                 /** 4.给PoolEntry添加定时任务,当PoolEntry对象达到最大生命周期时间后触发定时任务将连接标记为被抛弃 */
14                 poolEntry.setFutureEol(houseKeepingExecutorService.schedule(
15                         () -> {
16                             /** 5.达到最大生命周期,抛弃连接 */
17                             if (softEvictConnection(poolEntry, "(connection has passed maxLifetime)", false /* not owner */)) {
18                                 /** 6.丢弃一个连接之后,调用addBagItem补充新的PoolEntry对象 */
19                                 addBagItem(connectionBag.getWaitingThreadCount());
20                             }
21                         },
22                         lifetime, MILLISECONDS));
23             }
24 
25             return poolEntry;
26         }
27         /** 异常捕获*/
28         catch (ConnectionSetupException e) {
29             if (poolState == POOL_NORMAL) { // we check POOL_NORMAL to avoid a flood of messages if shutdown() is running concurrently
30                 logger.error("{} - Error thrown while acquiring connection from data source", poolName, e.getCause());
31                 lastConnectionFailure.set(e);
32             }
33             return null;
34         }
35         catch (SQLException e) {
36             if (poolState == POOL_NORMAL) { // we check POOL_NORMAL to avoid a flood of messages if shutdown() is running concurrently
37                 logger.debug("{} - Cannot acquire connection from data source", poolName, e);
38                 lastConnectionFailure.set(new ConnectionSetupException(e));
39             }
40             return null;
41         }
42         catch (Exception e) {
43             if (poolState == POOL_NORMAL) { // we check POOL_NORMAL to avoid a flood of messages if shutdown() is running concurrently
44                 logger.error("{} - Error thrown while acquiring connection from data source", poolName, e);
45                 lastConnectionFailure.set(new ConnectionSetupException(e));
46             }
47             return null;
48         }
49     }

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

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

1 void recycle(final PoolEntry poolEntry)
2 {
3       metricsTracker.recordConnectionUsage(poolEntry);
4       /** 调用ConcurrentBag的回收方法 */
5       connectionBag.requite(poolEntry);
6 }

 1 /** 回收元素方法 */
 2     public void requite(final T bagEntry)
 3     {
 4         /** 1.设置状态为未使用 */
 5         bagEntry.setState(STATE_NOT_IN_USE);
 6 
 7         /** 2.如果当前存在等待线程,则优先将元素给等待线程 */
 8         for (int i = 0; waiters.get() > 0; i++) {
 9             /** 2.1.将元素添加到*阻塞队列中,等待其他线程获取 */
10             if (bagEntry.getState() != STATE_NOT_IN_USE || handoffQueue.offer(bagEntry)) {
11                 return;
12             }
13             else if ((i & 0xff) == 0xff) {
14                 parkNanos(MICROSECONDS.toNanos(10));
15             }
16             else {
17                 /** 当前线程不再继续执行 */
18                 yield();
19             }
20         }
21         /** 3.如果当前连接没有被其他线程使用,则添加到当前线程的ThreadLocal中 */
22         final List<Object> threadLocalList = threadList.get();
23         if (threadLocalList.size() < 50) {
24             threadLocalList.add(weakThreadLocals ? new WeakReference<>(bagEntry) : bagEntry);
25         }
26     }