连接池——数据库连接池

什么是连接池

  连接池是创建和管理一个连接的缓冲池的技术，这些连接准备好被任何需要它们的线程使用。

为什么使用连接池

  在使用客户端进行连接服务端网络通信时，大部分是基于 TCP 连接。以数据库 MySQL 为例，我们通过Wireshark抓包整个连接过程分析下。

  第一部分是前三个数据包。可以看出这是一个 TCP 的三次握手过程；第二部分是 MySQL 服务端校验客户端账号的过程。

  所以每个客户端在和服务端建立连接的时候， 需要做的事情就是 TCP握手、用户校验、 获取权限，整个连接过程还是非常耗时的。而对于连接池来说，作用就是避免频繁创建连接和销毁连接。

连接池功能

  连接池一般对外提供获得连接、归还连接的接口给客户端使用，并可以配置连接数等参数，在内部则实现连接建立、连接管理、空闲连接回收、连接可用性检测等功能

连接池

Hikari数据库连接池

  数据库连接池和线程池一样，都属于池化资源。当需要执行 SQL 时，并不是直接创建一个数据 库连接，而是从连接池中获取一个；当 SQL 执行完，也并不是将数据库连接真的关掉，而是将其归还到连接池中。我们以HikariCP (源码版本5.0.1) 连接池为例，从连接的获取、归还、关闭、创建几个方面详细介绍HikariCP生命周期里的那些事。

获取连接

  获取连接是HikariCP的核心功能,我们先看下大概的执行过程，大概就是去连接池获取可用连接，没有则创建。

  接着我们看下具体的源码实现，首先HikariDataSource对象通过调用getConnection()方法获取连接，这里主要是校验连接以及连接池的初始化

//HikariDataSource#getConnection()
private final AtomicBoolean isShutdown = new AtomicBoolean();  
private final HikariPool fastPathPool;  
//这里使用了volatile
private volatile HikariPool pool;
public Connection getConnection() throws SQLException {
  //检查连接是否已经关闭
  if (isClosed()) {
     throw new SQLException("HikariDataSource "   this   " has been closed.");
  }
  //连接池不为空直接返回连接
  if (fastPathPool != null) {
     return fastPathPool.getConnection();
  }
  HikariPool result = pool;
  if (result == null) {
     //双重校验锁
     synchronized (this) {
        result = pool;
        if (result == null) {
           //参数校验
           validate();
           LOGGER.info("{} - Starting...", getPoolName());
           try {
              //初始化连接池
              pool = result = new HikariPool(this);
              this.seal();
           } catch (PoolInitializationException pie) {
              if (pie.getCause() instanceof SQLException) {
                 throw (SQLException) pie.getCause();
              } else {
                 throw pie;
              }
           }
           LOGGER.info("{} - Start completed.", getPoolName());
        }
     }
  }
  //连接池返回连接
  return result.getConnection();
}

  接着我们是HikariPool里面的getConnection()方法，HikariPool内部通过ConcurrentBag（并发容器）的borrow 方法获取。最后通过创建一个物理连接并返回其代理连接Proxy Connection。

//HikariPool#getConnection
public Connection getConnection(final long hardTimeout) throws SQLException{
  //获取信号量，没有阻塞等待
  suspendResumeLock.acquire();
  final var startTime = currentTime();
  try {
     var timeout = hardTimeout;
     do {
        //借用连接
        var poolEntry = connectionBag.borrow(timeout, MILLISECONDS);
        if (poolEntry == null) {
           break;
        }
        final var now = currentTime();
        if (poolEntry.isMarkedEvicted() || (elapsedMillis(poolEntry.lastAccessed, now) > aliveBypassWindowMs && isConnectionDead(poolEntry.connection))) {
           closeConnection(poolEntry, poolEntry.isMarkedEvicted() ? EVICTED_CONNECTION_MESSAGE : DEAD_CONNECTION_MESSAGE);
           timeout = hardTimeout - elapsedMillis(startTime);
        } else {
           //借用的连接没有过期，通过代理创建连接
           metricsTracker.recordBorrowStats(poolEntry, startTime);
           return poolEntry.createProxyConnection(leakTaskFactory.schedule(poolEntry));
        }
     } while (timeout > 0L);
     // 超时抛出异常
     metricsTracker.recordBorrowTimeoutStats(startTime);
     throw createTimeoutException(startTime);
  } catch (InterruptedException e) {
     Thread.currentThread().interrupt();
     throw new SQLException(poolName   " - Interrupted during connection acquisition", e);
  }  finally {
     suspendResumeLock.release();
  }
}

  最后我们看下ConcurrentBag#borrow方法，这里就是具体的从连接池里面获取连接。它将首先尝试从线程的ThreadLocal最近使用的连接列表中获取未使用的连接。再去共享连接池中获取。

//ConcurrentBag#borrow
private final ThreadLocal<ArrayList<WeakReference<IConcurrentBagEntry>>> threadList;
// 用于存储所有的数据库连接
protected final CopyOnWriteArrayList<T> sharedList;
public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException{
  //它将首先尝试从线程的`ThreadLocal`最近使用的连接列表中获取
  final var list = threadList.get();
  for (int i = list.size() - 1; i >= 0; i--) {
     final var entry = list.remove(i);
     final T bagEntry = weakThreadLocals ? ((WeakReference<T>) entry).get() : (T) entry;
     // 线程本地存储中的连接也可以被窃取，需要使用CAS修改状态为使用中
     if (bagEntry != null && bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
        return bagEntry;
     }
  }
  final int waiting = waiters.incrementAndGet();
  try {
     //线程本地存储中无空闲连接，则从共享队列中获取
     for (T bagEntry : sharedList) {
        if (bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
           if (waiting > 1) {
           // 如果我们已经窃取了另一个等待着的连接，那么会新增另一个连接
              listener.addBagItem(waiting - 1);
           }
           // 如果共享队列中有空闲连接，则返回
           return bagEntry;
        }
     }
     //添加连接监听，后面通过线程池异步创建连接
     listener.addBagItem(waiting);
     // 共享队列中没有连接，则需要等待
     timeout = timeUnit.toNanos(timeout);
     do {
        final var start = currentTime();
        final T bagEntry = handoffQueue.poll(timeout, NANOSECONDS);
        if (bagEntry == null || bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
           return bagEntry;
        }
        // 重新计算等待时间
        timeout -= elapsedNanos(start);
     } while (timeout > 10_000);
     // 超时没有获取到连接，返回 null
     return null;
  }
  finally {
     waiters.decrementAndGet();
  }
}

归还连接

  连接的归还和连接的借用是两个大致相反的过程。主要就是设置使用状态并且把归还的连接放入到相对应的ThreadLocal列表里。

  在上面获取连接是通过代理方式创建，进行归还时也是调用了代理层方法。也就是封装在ProxyConnection代理连接中的close方法

//ProxyConnection#close
public final void close() throws SQLException{
    // 关闭所有打开的Statement
    closeStatements();
    if (delegate != ClosedConnection.CLOSED_CONNECTION) {
         leakTask.cancel();
         try {
            if (isCommitStateDirty && !isAutoCommit) {
                 // 如果存在脏提交或者没有自动提交，则连接回滚
                 delegate.rollback();
                 LOGGER.debug("{} - Executed rollback on connection {} due to dirty commit state on close().", poolEntry.getPoolName(), delegate);
            }
            if (dirtyBits != 0) {
                 //重新设置连接属性
                 poolEntry.resetConnectionState(this, dirtyBits);
            }
            delegate.clearWarnings();
         }
         catch (SQLException e) {
            //连接中止，抛出异常
            if (!poolEntry.isMarkedEvicted()) {
                 throw checkException(e);
            }
         }
         finally {
            delegate = ClosedConnection.CLOSED_CONNECTION;
            //释放连接
            poolEntry.recycle();
         }
    }
}

  HikariPool中的recycle方法，最后实际调用的还是ConcurrentBag##requite方法，最后我们再看下这个方法。

//ConcurrentBag#requite
public void requite(final T bagEntry) {
    // 更新连接状态
    bagEntry.setState(STATE_NOT_IN_USE);
    // 如果有等待的线程，则直接分配给线程，无需进入任何队列
    for (var i = 0; waiters.get() > 0; i  ) {
        if (bagEntry.getState() != STATE_NOT_IN_USE || handoffQueue.offer(bagEntry)) {
            return;
        }
        else if ((i & 0xff) == 0xff) {
            parkNanos(MICROSECONDS.toNanos(10));
        }
        else {
            Thread.yield();
        }
    }
    // 如果没有等待的线程，把归还的连接放入到相对应的ThreadLocal里
    final var threadLocalList = threadList.get();
    if (threadLocalList.size() < 50) {
        threadLocalList.add(weakThreadLocals ? new WeakReference<>(bagEntry) : bagEntry);
    }
}

关闭连接

  这里的关闭连接指的是关闭了空闲连接，中止等活动连接。并非是直接关闭数据库连接池。关闭连接就是移除连接池中的连接，并且调用JDBC方法关闭真实的物理连接。在上面的归还连接的时候，如果连接断开，就会调用关闭连接的方法。它会关闭真实的底层连接。

  当HikariPool执行closeConnection方法时,首先从ConcurrentBag中移除PoolEntry

//HikariPool#closeConnection
void closeConnection(final PoolEntry poolEntry, final String closureReason){
    //ConcurrentBag中移除PoolEntry
    if (connectionBag.remove(poolEntry)) {
        //关闭poolEntry
        final var connection = poolEntry.close();
        closeConnectionExecutor.execute(() -> {
            //独立线程池进行物理连接的关闭
            quietlyCloseConnection(connection, closureReason);
            if (poolState == POOL_NORMAL) {
                fillPool(false);
            }
        });
    }
}

  然后PoolEntry自身关闭，并且所有的数据库连接共享列表移除连接。

//ConcurrentBag#remove 
public boolean remove(final T bagEntry) {
      //CAS设置关闭状态
      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);
      }
      //从线程本地存储中删除连接
      threadList.get().remove(bagEntry);

      return removed;
}

  接着独立线程池 closeConnectionExecutor (本质是ThreadPoolExecutor）调用 JDBC 的方法进行物理连接的关闭。

 //PoolBase#quietlyCloseConnection
   void quietlyCloseConnection(final Connection connection, final String closureReason){
      if (connection != null) {
         try {
            logger.debug("{} - Closing connection {}: {}", poolName, connection, closureReason);
            try (connection) {
               setNetworkTimeout(connection, SECONDS.toMillis(15));
            } catch (SQLException e) {
            }
         }
         catch (Exception e) {
            logger.debug("{} - Closing connection {} failed", poolName, connection, e);
         }
      }
   }

创建连接

  其实在上面获取连接的时候如果空闲列表没有可用连接就会添加创建连接，或者填充最小空闲连接数的时候，会去创建连接。

  在上面ConcurrentBag的borrrow也就是获取连接的时候，分别执行了listener.addBagltem(waiting-1)和listener.addBagItem(waiting)

  也就是如果空闲列表没有可用连接就会添加连接监听，然后通过独立的线程池 addConnectionExecutor去创建对应的物理的连接,创建完成以后的连接会被封装为PoolEntry并放入ConcurrentBag。

//PoolEntryCreator#call
public Boolean call() {
    var backoffMs = 10L;
    var added = false;
    try {
        //基于最大连接数与状态判断连接池是否需要添加连接
        while (shouldContinueCreating()) {
            // 创建PoolEntry
            final var poolEntry = createPoolEntry();
            if (poolEntry != null) {
                added = true;
                backoffMs = 10L;
                connectionBag.add(poolEntry);
                logger.debug("{} - Added connection {}", poolName, poolEntry.connection);
            } else {  
                //创建失败，休眠10ms，然后重试
                backoffMs = Math.min(SECONDS.toMillis(5), backoffMs * 2);
                if (loggingPrefix != null)
                    logger.debug("{} - Connection add failed, sleeping with backoff: {}ms", poolName, backoffMs);
            }

            quietlySleep(backoffMs);
        }
    }
    finally {
        addConnectionQueueDepth.decrementAndGet();
        if (added && loggingPrefix != null) logPoolState(loggingPrefix);
    }
    return Boolean.FALSE;
}

  addConnectionExecutor线程调用HikariPool的createPoolEntry方法进行连接生成，PoolBase提供的 newPoolEntry会先进行物理连接的创建，最后我们看下里面的关键createPoolEntry方法

 //HikariPool#createPoolEntry
private PoolEntry createPoolEntry() {
    try {
        //调用JDBC的DriverManager获取连接
        final var poolEntry = newPoolEntry();

        final var maxLifetime = config.getMaxLifetime();
        if (maxLifetime > 0) {
            // 在maxLifetime的基础上减去一个随机数，防止同一时间大量连接被关闭
            final var variance = maxLifetime > 10_000 ? ThreadLocalRandom.current().nextLong( maxLifetime / 40 ) : 0;
            final var lifetime = maxLifetime - variance;
            // 连接超过MaxLifeTime后，重新创建连接
            poolEntry.setFutureEol(houseKeepingExecutorService.schedule(new MaxLifetimeTask(poolEntry), lifetime, MILLISECONDS));
        }
        final long keepaliveTime = config.getKeepaliveTime();
        if (keepaliveTime > 0) {
            final var variance = ThreadLocalRandom.current().nextLong(keepaliveTime / 10);
            final var heartbeatTime = keepaliveTime - variance;
            poolEntry.setKeepalive(houseKeepingExecutorService.scheduleWithFixedDelay(new KeepaliveTask(poolEntry), heartbeatTime, heartbeatTime, MILLISECONDS));
        }

        return poolEntry;
    }
    catch (ConnectionSetupException e) {
        if (poolState == POOL_NORMAL) { 
            logger.error("{} - Error thrown while acquiring connection from data source", poolName, e.getCause());
            lastConnectionFailure.set(e);
        }
    }
    catch (Exception e) {
        if (poolState == POOL_NORMAL) { 
            logger.debug("{} - Cannot acquire connection from data source", poolName, e);
        }
    }

    return null;
}

数据库连接池的配置

  连接池提供了许多参数，最重要的参数是最大连接数，最大连接数不是设置得越大越好。对一些人来说可能是违反直觉的。你有一个网站，通常仍有10000名用户同时请求数据库，每秒约有20000笔交易。你觉得的连接池应该有多大？

  可以在下面的PostgreSQL基准测试中看到，TPS速率在大约50个连接时开始趋于平缓

  下面的公式是由PostgreSQL提供，但将在很大程度上适用于所有数据库。最好的是应该测试应用程序，即模拟预期负载，并围绕此起点尝试不同的连接数配置。

$connections=((coreCount\ast 2)effectiveSpindleCount)$

coreCount = 核心数。effectiveSpindleCount = 有效磁盘数。如果活跃数据全部被缓存了，那么有效磁盘数是0，随着缓存命中率的下降，有效磁盘数逐渐趋近于实际的磁盘数。 参考：github.com/brettwooldr…

  那这意味着什么？

  一个8核数据库服务器的连接池大小 ((8 * 2) 1) = 17，差不多20左右，大家感兴趣的可以基于这个公式配置进行负载测试试下。所以不要过度配置数据库。

参考

• github.com/brettwooldr…

• github.com/brettwooldr…

这篇好文章是转载于：学新通技术网