DB连接池HikariCP为什么如此快

Posted 2021-04-06 攻城狮DermanYuan

1、背景介绍

我们做过的项目中，只要连接数据库，就不可避免的使用数据库连接池，而且面试的时候，数据库肯定会被问到的。说到数据库就会问到连接池，大部分的业务码工，只是配置下参数，再问其他一脸茫然。作为想往更高处走的码工，除了知道怎么配置还要知道参数的意义及连接池原理，今天带你走进db连接池HikariCP的源码世界，一点一点解开她神秘面纱。

2、DB连接池HikariCP

JDBC的连接池的实现并不复杂，主要是对JDBC中几个核心功能Connection、Connection.close、Statement、PreparedStatment、CallableStatement以及ResultSet的封装及优化。下面来看看HikariCP官网的性能测试对比

从上述结果看到，HikariCP的性能甩其他c3p0、tomcat-jdbc连接池好几条街。以致后来BoneCP作者都放弃了维护，在Github项目主页推荐大家使用HikariCP。另外，Spring Boot将在2.0版本中把HikariCP作为其默认的JDBC连接池。

需要指出的是，上图中的测试数据，是HikariCP作者对各个连接池的DataSource.getConnection()/Connection.close()和Connection.prepareStatement()/Statement.execute()/Statement.close()方法的性能测试结果。

下面从这几个核心功能去分析源码，HikariCP为什么这么快，分析的版本2.5.1。

3、通过配置引入hikari数据源

<bean id="masterDataSource" class="com.zaxxer.hikari.HikariDataSource" destroy-method="close"> <property name="driverClassName" value="${default.jdbc.driverClassName}" /> <property name="jdbcUrl" value="${master.jdbcurl}" /> <property name="username" value="${master.username}" /> <property name="password" value="${master.password}" /> <property name="minimumIdle" value="20" /> <property name="maximumPoolSize" value="20" /> <property name="connectionTimeout" value="250" /></bean>

4、HikariDataSource分析

该类继承了HikariConfig，声明了HikariPool 连接池。HikariConfig保存了连接池的配置信息，HikariPool保存了连接的池子，提供了连接/关闭/指标监控等基本功能。

要操作DB，首先获取connection。获取连接的操作很简单，校验数据源是否关闭，然后获取连接池，从池子中拿到连接，源码如下：

public Connection getConnection() throws SQLException{ //校验数据源是否关闭 if (isClosed()) { throw new SQLException("HikariDataSource " + this + " has been closed."); }
 if (fastPathPool != null) { return fastPathPool.getConnection(); }
 // See http://en.wikipedia.org/wiki/Double-checked_locking#Usage_in_Java HikariPool result = pool; if (result == null) { synchronized (this) { result = pool; if (result == null) {           //校验参数是否合法及赋默认值 validate(); LOGGER.info("{} - Started.", getPoolName());           //初始化连接池，连接池的数据源、配置信息、创建池子、建立连接等操作。 pool = result = new HikariPool(this); } } }  //从连接池中返回一个连接，提现了对连接池的管理维护机制 return result.getConnection();}

获取连接的关键步骤是获取连接池，看看HikariCP的连接池初始化过程，源码如下：

/*** Construct a HikariPool with the specified configuration.** @param config a HikariConfig instance*/public HikariPool(final HikariConfig config){  //父类初始化配置信息和数据源，这一步很重要。 super(config);  //初始化连接池的实体对象，这一步是连接池的核心实现。 this.connectionBag = new ConcurrentBag<>(this);  .....  //注册监控信息 registerMBeans(this);
 ThreadFactory threadFactory = config.getThreadFactory(); this.addConnectionExecutor = createThreadPoolExecutor(config.getMaximumPoolSize(), poolName + " connection adder", threadFactory, new ThreadPoolExecutor.DiscardPolicy()); this.closeConnectionExecutor = createThreadPoolExecutor(config.getMaximumPoolSize(), poolName + " connection closer", threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());
 if (config.getScheduledExecutorService() == null) { threadFactory = threadFactory != null ? threadFactory : new DefaultThreadFactory(poolName + " housekeeper", true); this.houseKeepingExecutorService = new ScheduledThreadPoolExecutor(1, threadFactory, new ThreadPoolExecutor.DiscardPolicy()); this.houseKeepingExecutorService.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); this.houseKeepingExecutorService.setRemoveOnCancelPolicy(true); } else { this.houseKeepingExecutorService = config.getScheduledExecutorService(); }
 this.leakTask = new ProxyLeakTask(config.getLeakDetectionThreshold(), houseKeepingExecutorService);
 this.houseKeepingExecutorService.scheduleWithFixedDelay(new HouseKeeper(), 100L, HOUSEKEEPING_PERIOD_MS, MILLISECONDS);}

4.1 初始化PoolBase实体及数据源

PoolBase(final HikariConfig config){ this.config = config;
 ....
 this.poolName = config.getPoolName(); this.connectionTimeout = config.getConnectionTimeout(); this.validationTimeout = config.getValidationTimeout(); this.lastConnectionFailure = new AtomicReference<>();
 initializeDataSource();}
//初始化数据源/*** Create/initialize the underlying DataSource.** @return a DataSource instance*/private void initializeDataSource(){ final String jdbcUrl = config.getJdbcUrl(); final String username = config.getUsername(); final String password = config.getPassword(); final String dsClassName = config.getDataSourceClassName(); final String driverClassName = config.getDriverClassName(); final Properties dataSourceProperties = config.getDataSourceProperties();
 DataSource dataSource = config.getDataSource(); if (dsClassName != null && dataSource == null) { dataSource = createInstance(dsClassName, DataSource.class); PropertyElf.setTargetFromProperties(dataSource, dataSourceProperties); } else if (jdbcUrl != null && dataSource == null) { //根据工程中xml配置信息会走到这里，初始化数据源 dataSource = new DriverDataSource(jdbcUrl, driverClassName, dataSourceProperties, username, password); }
 if (dataSource != null) { setLoginTimeout(dataSource); createNetworkTimeoutExecutor(dataSource, dsClassName, jdbcUrl); }
 this.dataSource = dataSource;}

4.2初始化连接池实体ConcurrentBag

可以看到最终的连接池放到了FastList集合，这个集合存放着IConcurrentBagEntry对象。

private final QueuedSequenceSynchronizer synchronizer;private final CopyOnWriteArrayList<T> sharedList;//连接池的最终实体，通过ThreadLocal实现lock-less机制，减少锁竞争提高并发性private final ThreadLocal<List<Object>> threadList;
/*** Construct a ConcurrentBag with the specified listener.** @param listener the IBagStateListener to attach to this bag*/public ConcurrentBag(final IBagStateListener listener){ this.listener = listener; this.weakThreadLocals = useWeakThreadLocals();
 this.waiters = new AtomicInteger(); this.sharedList = new CopyOnWriteArrayList<>(); this.synchronizer = new QueuedSequenceSynchronizer(); if (weakThreadLocals) { this.threadList = new ThreadLocal<>(); } else { //根据工程中xml配置信息会走到这里，初始化线程列表 this.threadList = new ThreadLocal<List<Object>>() { @Override protected List<Object> initialValue() { return new FastList<>(IConcurrentBagEntry.class, 16); } }; }}

5、ConcurrentBag：更好的并发集合类

ConcurrentBag的实现是一个专门为连接池设计的lock-less集合，实现了比LinkedBlockingQueue、LinkedTransferQueue更好的并发性能。

ConcurrentBag内部同时使用了ThreadLocal和CopyOnWriteArrayList来存储元素，其中CopyOnWriteArrayList是线程共享的。ConcurrentBag采用了queue-stealing的机制获取元素：首先尝试从ThreadLocal中获取属于当前线程的元素来避免锁竞争，如果没有可用元素则再次从共享的CopyOnWriteArrayList中获取，进而减少伪共享(false sharing)的发生。

6、FastList替换ArrayList

FastList只实现了List接口一些基本功能，例如：add,get,remove,size等。在get方法中去掉了不必要的range checking，只要保证索引合法那么rangeCheck就成为了不必要的计算开销。

ArrayList的remove(Object)方法是从头开始遍历数组，而FastList是从数组的尾部开始遍历，因此更为高效。所以，HikariCP使用List来保存打开的Statement，当Statement关闭或Connection关闭时需要将对应的Statement从FastList中移除。通常情况下，同一个Connection创建了多个Statement时，会按照先创建后关闭的顺序执行。

7、从连接池中“租借”连接

连接池经过一系列的初始化后，池子已经搭建完毕，接下来从池子中获取连接，源码如下：

/*** Get a connection from the pool, or timeout after the specified number of milliseconds.** @param hardTimeout the maximum time to wait for a connection from the pool* @return a java.sql.Connection instance* @throws SQLException thrown if a timeout occurs trying to obtain a connection*/public final Connection getConnection(final long hardTimeout) throws SQLException{  //又是作者自己实现的挂起恢复锁，根据框架的需要对semphere的封装 suspendResumeLock.acquire(); final long startTime = clockSource.currentTime();
 try { long timeout = hardTimeout; do { //从connectionBag租借连接，如果租借不到则抛异常   //在租借的过程中，会发生 很多事情，比如：创建连接，连接的过期或Idel时间校验 final PoolEntry poolEntry = connectionBag.borrow(timeout, MILLISECONDS); if (poolEntry == null) { break; // We timed out... break and throw exception }
 final long now = clockSource.currentTime(); if (poolEntry.isMarkedEvicted() || (clockSource.elapsedMillis(poolEntry.lastAccessed, now) > ALIVE_BYPASS_WINDOW_MS && !isConnectionAlive(poolEntry.connection))) { closeConnection(poolEntry, "(connection is evicted or dead)"); // Throw away the dead connection (passed max age or failed alive test) timeout = hardTimeout - clockSource.elapsedMillis(startTime); } else {       //租借的连接，没过期也没被抛弃则设置监控指标，通过ProxyFactory.getProxyConnection创建连接代理（最核心的地方） metricsTracker.recordBorrowStats(poolEntry, startTime); return poolEntry.createProxyConnection(leakTask.schedule(poolEntry), now); } } while (timeout > 0L); } catch (InterruptedException e) { throw new SQLException(poolName + " - Interrupted during connection acquisition", e); } finally { suspendResumeLock.release(); }
 throw createTimeoutException(startTime);}

/*** The method will borrow a BagEntry from the bag, blocking for the* specified timeout if none are available.** @param timeout how long to wait before giving up, in units of unit* @param timeUnit a <code>TimeUnit</code> determining how to interpret the timeout parameter* @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{ // Try the thread-local list first List<Object> list = threadList.get(); if (weakThreadLocals && list == null) { list = new ArrayList<>(16); threadList.set(list); }
 for (int i = list.size() - 1; i >= 0; i--) { //这里的list是FastList，remove数据时从末尾删除 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)) {   //1.优先从本线程的ThreadLocal中拿连接，如果拿到则直接返回 return bagEntry; } }
 //2.ThreadLocal中没有租借到连接，则从sharedList获取 // Otherwise, scan the shared list ... for maximum of timeout timeout = timeUnit.toNanos(timeout); Future<Boolean> addItemFuture = null; final long startTime = System.nanoTime(); final long originTimeout = timeout; long startSeq; 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... if (waiters.get() > 1 && addItemFuture == null) { listener.addBagItem(); }
 return bagEntry; } } } while (startSeq < synchronizer.currentSequence());
 if (addItemFuture == null || addItemFuture.isDone()) { addItemFuture = listener.addBagItem(); }
 timeout = originTimeout - (System.nanoTime() - startTime); } while (timeout > 10_000L && synchronizer.waitUntilSequenceExceeded(startSeq, timeout)); } finally { waiters.decrementAndGet(); }}

租借到PoolEntry后，开始创建代理连接，poolEntry.createProxyConnection。打开代理工厂，看到方法里面都是throw异常的代码，注释中：Body is replaced (injected) by JavassistProxyFactory，原来HikariCP是通过Javassit动态代理在编译的时候植入字节码，从而达到很高的性能，源码如下：

static ProxyConnection getProxyConnection(final PoolEntry poolEntry, final Connection connection, final FastList<Statement> openStatements, final ProxyLeakTask leakTask, final long now, final boolean isReadOnly, final boolean isAutoCommit){ // Body is replaced (injected) by JavassistProxyFactory throw new IllegalStateException("You need to run the CLI build and you need target/classes in your classpath to run.");}
static Statement getProxyStatement(final ProxyConnection connection, final Statement statement){ // Body is replaced (injected) by JavassistProxyFactory throw new IllegalStateException("You need to run the CLI build and you need target/classes in your classpath to run.");}
static CallableStatement getProxyCallableStatement(final ProxyConnection connection, final CallableStatement statement){ // Body is replaced (injected) by JavassistProxyFactory throw new IllegalStateException("You need to run the CLI build and you need target/classes in your classpath to run.");}.....

8、动态代理JDBC核心方法的字节码

HikariCP利用了一个第三方的Java字节码修改类库Javassist来生成委托实现动态代理。动态代理的实现在JavassistProxyFactory类，源码如下：

private static ClassPool classPool;
 public static void main(String... args) { classPool = new ClassPool(); classPool.importPackage("java.sql"); classPool.appendClassPath(new LoaderClassPath(JavassistProxyFactory.class.getClassLoader()));
 try { // Cast is not needed for these String methodBody = "{ try { return delegate.method($$); } catch (SQLException e) { throw checkException(e); } }"; generateProxyClass(Connection.class, ProxyConnection.class.getName(), methodBody); generateProxyClass(Statement.class, ProxyStatement.class.getName(), methodBody); generateProxyClass(ResultSet.class, ProxyResultSet.class.getName(), methodBody);
 // For these we have to cast the delegate methodBody = "{ try { return ((cast) delegate).method($$); } catch (SQLException e) { throw checkException(e); } }"; generateProxyClass(PreparedStatement.class, ProxyPreparedStatement.class.getName(), methodBody); generateProxyClass(CallableStatement.class, ProxyCallableStatement.class.getName(), methodBody);
 modifyProxyFactory(); } catch (Exception e) { throw new RuntimeException(e); } }

private static void modifyProxyFactory() throws Exception { System.out.println("Generating method bodies for com.zaxxer.hikari.proxy.ProxyFactory");
 String packageName = ProxyConnection.class.getPackage().getName(); CtClass proxyCt = classPool.getCtClass("com.zaxxer.hikari.pool.ProxyFactory"); for (CtMethod method : proxyCt.getMethods()) { switch (method.getName()) { case "getProxyConnection": method.setBody("{return new " + packageName + ".HikariProxyConnection($$);}"); break; case "getProxyStatement": method.setBody("{return new " + packageName + ".HikariProxyStatement($$);}"); break; case "getProxyPreparedStatement": method.setBody("{return new " + packageName + ".HikariProxyPreparedStatement($$);}"); break; case "getProxyCallableStatement": method.setBody("{return new " + packageName + ".HikariProxyCallableStatement($$);}"); break; case "getProxyResultSet": method.setBody("{return new " + packageName + ".HikariProxyResultSet($$);}"); break; default: // unhandled method break; } }
 proxyCt.writeFile("target/classes"); }

9、总结

来到这里，HikariCP的核心源码已经看的差不多了，咱们来梳理下整个流程，只有清楚整个流程怎么调用，才能做到心中有数。

工程中通过配置文件配置HikariCP数据源(HikariDataSource)-->初始化数据源配置信息(HikariConfig)-->初始化连接池(HikariPool)-->初始化数据源(dataSource)-->构造连接池包实体(ConcurrentBag)-->初始化线程列表(threadList，根据FastList初始化IConcurrentBagEntry类型的列表)-->注册健康检测(setHealthCheckRegistry)-->注册指标监控(registerMBeans)-->返回连接池(HikariPool)对象-->获取连接(getConnection)-->从connectionBag租借连接-->返回连接代理(proxyConnection)-->通过javassist动态生成字节码(connection, statement等)。

性能高的关键点：FastList替换ArrayList，自定义的ConcurrentBag连接池实体，javassit动态生成JDBC核心方法的代理字节码。

如果感觉有收获，欢迎关注 转发 点好看，谢谢~ 我是 DermanYuan