【追光者系列】HikariCP源码分析之evict、时钟回拨、连接创建生命周期

private volatile boolean evict;boolean isMarkedEvicted() {      return evict;   }   void markEvicted() {      this.evict = true;   }

 /**    * 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 Connection getConnection(final long hardTimeout) throws SQLException {      suspendResumeLock.acquire();      final long startTime = currentTime();      try {         long timeout = hardTimeout;         do {            PoolEntry poolEntry = connectionBag.borrow(timeout, MILLISECONDS);            if (poolEntry == null) {               break; // We timed out... break and throw exception            }            final long now = currentTime();            if (poolEntry.isMarkedEvicted() || (elapsedMillis(poolEntry.lastAccessed, now) > ALIVE_BYPASS_WINDOW_MS && !isConnectionAlive(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), now);            }         } 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();      }   }

if (poolEntry.isMarkedEvicted() || (elapsedMillis(poolEntry.lastAccessed, now) > ALIVE_BYPASS_WINDOW_MS && !isConnectionAlive(poolEntry.connection))) {               closeConnection(poolEntry, poolEntry.isMarkedEvicted() ? EVICTED_CONNECTION_MESSAGE : DEAD_CONNECTION_MESSAGE);               timeout = hardTimeout - elapsedMillis(startTime);            }

 private static final String EVICTED_CONNECTION_MESSAGE = "(connection was evicted)";   private static final String DEAD_CONNECTION_MESSAGE = "(connection is dead)";   /**    * Permanently close the real (underlying) connection (eat any exception).    *    * @param poolEntry poolEntry having the connection to close    * @param closureReason reason to close    */   void closeConnection(final PoolEntry poolEntry, final String closureReason) {      if (connectionBag.remove(poolEntry)) {         final Connection connection = poolEntry.close();         closeConnectionExecutor.execute(() -> {            quietlyCloseConnection(connection, closureReason);            if (poolState == POOL_NORMAL) {               fillPool();            }         });      }   }

// ***********************************************************************   //                           JDBC methods   // ***********************************************************************   void quietlyCloseConnection(final Connection connection, final String closureReason) {      if (connection != null) {         try {            LOGGER.debug("{} - Closing connection {}: {}", poolName, connection, closureReason);            try {               setNetworkTimeout(connection, SECONDS.toMillis(15));            }            finally {               connection.close(); // continue with the close even if setNetworkTimeout() throws            }         }         catch (Throwable e) {            LOGGER.debug("{} - Closing connection {} failed", poolName, connection, e);         }      }   }

// ***********************************************************************   //                           Private methods   // ***********************************************************************   /**    * Creating new poolEntry.  If maxLifetime is configured, create a future End-of-life task with 2.5% variance from    * the maxLifetime time to ensure there is no massive die-off of Connections in the pool.    */   private PoolEntry createPoolEntry() {      try {         final PoolEntry poolEntry = newPoolEntry();         final long maxLifetime = config.getMaxLifetime();         if (maxLifetime > 0) {            // variance up to 2.5% of the maxlifetime            final long variance = maxLifetime > 10_000 ? ThreadLocalRandom.current().nextLong( maxLifetime / 40 ) : 0;            final long lifetime = maxLifetime - variance;            poolEntry.setFutureEol(houseKeepingExecutorService.schedule(               () -> {                  if (softEvictConnection(poolEntry, "(connection has passed maxLifetime)", false /* not owner */)) {                     addBagItem(connectionBag.getWaitingThreadCount());                  }               },               lifetime, MILLISECONDS));         }         return poolEntry;      }      catch (Exception e) {         if (poolState == POOL_NORMAL) { // we check POOL_NORMAL to avoid a flood of messages if shutdown() is running concurrently            LOGGER.debug("{} - Cannot acquire connection from data source", poolName, (e instanceof ConnectionSetupException ? e.getCause() : e));         }         return null;      }   }

/**    * Evict a Connection from the pool.    *    * @param connection the Connection to evict (actually a {@link ProxyConnection})    */   public void evictConnection(Connection connection) {      ProxyConnection proxyConnection = (ProxyConnection) connection;      proxyConnection.cancelLeakTask();      try {         softEvictConnection(proxyConnection.getPoolEntry(), "(connection evicted by user)", !connection.isClosed() /* owner */);      }      catch (SQLException e) {         // unreachable in HikariCP, but we're still forced to catch it      }   }

 /**    * "Soft" evict a Connection (/PoolEntry) from the pool.  If this method is being called by the user directly    * through {@link com.zaxxer.hikari.HikariDataSource#evictConnection(Connection)} then {@code owner} is {@code true}.    *    * If the caller is the owner, or if the Connection is idle (i.e. can be "reserved" in the {@link ConcurrentBag}),    * then we can close the connection immediately.  Otherwise, we leave it "marked" for eviction so that it is evicted    * the next time someone tries to acquire it from the pool.    *    * @param poolEntry the PoolEntry (/Connection) to "soft" evict from the pool    * @param reason the reason that the connection is being evicted    * @param owner true if the caller is the owner of the connection, false otherwise    * @return true if the connection was evicted (closed), false if it was merely marked for eviction    */   private boolean softEvictConnection(final PoolEntry poolEntry, final String reason, final boolean owner) {      poolEntry.markEvicted();      if (owner || connectionBag.reserve(poolEntry)) {         closeConnection(poolEntry, reason);         return true;      }      return false;   }

 /**    * The method is used to make an item in the bag "unavailable" for    * borrowing.  It is primarily used when wanting to operate on items    * returned by the <code>values(int)``` method.  Items that are    * reserved can be removed from the bag via <code>remove(T)```    * without the need to unreserve them.  Items that are not removed    * from the bag can be make available for borrowing again by calling    * the <code>unreserve(T)``` method.    *    * @param bagEntry the item to reserve    * @return true if the item was able to be reserved, false otherwise    */   public boolean reserve(final T bagEntry) {      return bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_RESERVED);   }

 public void softEvictConnections() {      connectionBag.values().forEach(poolEntry -> softEvictConnection(poolEntry, "(connection evicted)", false /* not owner */));   }

public void softEvictConnections() {      connectionBag.values().forEach(poolEntry -> softEvictConnection(poolEntry, "(connection evicted)", false /* not owner */));   }

/**    * "Soft" evict a Connection (/PoolEntry) from the pool.  If this method is being called by the user directly    * through {@link com.zaxxer.hikari.HikariDataSource#evictConnection(Connection)} then {@code owner} is {@code true}.    *    * If the caller is the owner, or if the Connection is idle (i.e. can be "reserved" in the {@link ConcurrentBag}),    * then we can close the connection immediately.  Otherwise, we leave it "marked" for eviction so that it is evicted    * the next time someone tries to acquire it from the pool.    *    * @param poolEntry the PoolEntry (/Connection) to "soft" evict from the pool    * @param reason the reason that the connection is being evicted    * @param owner true if the caller is the owner of the connection, false otherwise    * @return true if the connection was evicted (closed), false if it was merely marked for eviction    */   private boolean softEvictConnection(final PoolEntry poolEntry, final String reason, final boolean owner) {      poolEntry.markEvicted();      if (owner || connectionBag.reserve(poolEntry)) {         closeConnection(poolEntry, reason);         return true;      }      return false;   }

 /**    * The method is used to make an item in the bag "unavailable" for    * borrowing.  It is primarily used when wanting to operate on items    * returned by the <code>values(int)``` method.  Items that are    * reserved can be removed from the bag via <code>remove(T)```    * without the need to unreserve them.  Items that are not removed    * from the bag can be make available for borrowing again by calling    * the <code>unreserve(T)``` method.    *    * @param bagEntry the item to reserve    * @return true if the item was able to be reserved, false otherwise    */   public boolean reserve(final T bagEntry) {      return bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_RESERVED);   }

/**    * The house keeping task to retire and maintain minimum idle connections.    */   private final class HouseKeeper implements Runnable {      private volatile long previous = plusMillis(currentTime(), -HOUSEKEEPING_PERIOD_MS);      @Override      public void run()      {         try {            // refresh timeouts in case they changed via MBean            connectionTimeout = config.getConnectionTimeout();            validationTimeout = config.getValidationTimeout();            leakTaskFactory.updateLeakDetectionThreshold(config.getLeakDetectionThreshold());            final long idleTimeout = config.getIdleTimeout();            final long now = currentTime();            // Detect retrograde time, allowing +128ms as per NTP spec.            if (plusMillis(now, 128) < plusMillis(previous, HOUSEKEEPING_PERIOD_MS)) {               LOGGER.warn("{} - Retrograde clock change detected (housekeeper delta={}), soft-evicting connections from pool.",                           poolName, elapsedDisplayString(previous, now));               previous = now;               softEvictConnections();               return;            }            else if (now > plusMillis(previous, (3 * HOUSEKEEPING_PERIOD_MS) / 2)) {               // No point evicting for forward clock motion, this merely accelerates connection retirement anyway               LOGGER.warn("{} - Thread starvation or clock leap detected (housekeeper delta={}).", poolName, elapsedDisplayString(previous, now));            }            previous = now;            String afterPrefix = "Pool ";            if (idleTimeout > 0L && config.getMinimumIdle() < config.getMaximumPoolSize()) {               logPoolState("Before cleanup ");               afterPrefix = "After cleanup  ";               final List<PoolEntry> notInUse = connectionBag.values(STATE_NOT_IN_USE);               int toRemove = notInUse.size() - config.getMinimumIdle();               for (PoolEntry entry : notInUse) {                  if (toRemove > 0 && elapsedMillis(entry.lastAccessed, now) > idleTimeout && connectionBag.reserve(entry)) {                     closeConnection(entry, "(connection has passed idleTimeout)");                     toRemove--;                  }               }            }            logPoolState(afterPrefix);            fillPool(); // Try to maintain minimum connections         }         catch (Exception e) {            LOGGER.error("Unexpected exception in housekeeping task", e);         }      }   }

// Detect retrograde time, allowing +128ms as per NTP spec.            if (plusMillis(now, 128) < plusMillis(previous, HOUSEKEEPING_PERIOD_MS)) {               LOGGER.warn("{} - Retrograde clock change detected (housekeeper delta={}), soft-evicting connections from pool.",                           poolName, elapsedDisplayString(previous, now));               previous = now;               softEvictConnections();               return;            }  /**    * Return the specified opaque time-stamp plus the specified number of milliseconds.    *    * @param time an opaque time-stamp    * @param millis milliseconds to add    * @return a new opaque time-stamp    */   static long plusMillis(long time, long millis) {      return CLOCK.plusMillis0(time, millis);   }

/** * 自生成Id生成器. *  * <p> * 长度为64bit,从高位到低位依次为 * </p> *  * <pre> * 1bit   符号位  * 41bits 时间偏移量从2016年11月1日零点到现在的毫秒数 * 10bits 工作进程Id * 12bits 同一个毫秒内的自增量 * </pre> *  * <p> * 工作进程Id获取优先级: 系统变量{@code sjdbc.self.id.generator.worker.id} 大于 环境变量{@code SJDBC_SELF_ID_GENERATOR_WORKER_ID} * ,另外可以调用@{@code CommonSelfIdGenerator.setWorkerId}进行设置 * </p> *  * @author gaohongtao */@Getter@Slf4jpublic class CommonSelfIdGenerator implements IdGenerator {    public static final long SJDBC_EPOCH;//时间偏移量，从2016年11月1日零点开始    private static final long SEQUENCE_BITS = 12L;//自增量占用比特    private static final long WORKER_ID_BITS = 10L;//工作进程ID比特    private static final long SEQUENCE_MASK = (1 << SEQUENCE_BITS) - 1;//自增量掩码（最大值）    private static final long WORKER_ID_LEFT_SHIFT_BITS = SEQUENCE_BITS;//工作进程ID左移比特数（位数）    private static final long TIMESTAMP_LEFT_SHIFT_BITS = WORKER_ID_LEFT_SHIFT_BITS + WORKER_ID_BITS;//时间戳左移比特数（位数）    private static final long WORKER_ID_MAX_VALUE = 1L << WORKER_ID_BITS;//工作进程ID最大值    @Setter    private static AbstractClock clock = AbstractClock.systemClock();    @Getter    private static long workerId;//工作进程ID    static {        Calendar calendar = Calendar.getInstance();        calendar.set(2016, Calendar.NOVEMBER, 1);        calendar.set(Calendar.HOUR_OF_DAY, 0);        calendar.set(Calendar.MINUTE, 0);        calendar.set(Calendar.SECOND, 0);        calendar.set(Calendar.MILLISECOND, 0);        SJDBC_EPOCH = calendar.getTimeInMillis();        initWorkerId();    }    private long sequence;//最后自增量    private long lastTime;//最后生成编号时间戳，单位：毫秒    static void initWorkerId() {        String workerId = System.getProperty("sjdbc.self.id.generator.worker.id");        if (!Strings.isNullOrEmpty(workerId)) {            setWorkerId(Long.valueOf(workerId));            return;        }        workerId = System.getenv("SJDBC_SELF_ID_GENERATOR_WORKER_ID");        if (Strings.isNullOrEmpty(workerId)) {            return;        }        setWorkerId(Long.valueOf(workerId));    }    /**     * 设置工作进程Id.     *      * @param workerId 工作进程Id     */    public static void setWorkerId(final Long workerId) {        Preconditions.checkArgument(workerId >= 0L && workerId < WORKER_ID_MAX_VALUE);        CommonSelfIdGenerator.workerId = workerId;    }    /**     * 生成Id.     *      * @return 返回@{@link Long}类型的Id     */    @Override    public synchronized Number generateId() {    //保证当前时间大于最后时间。时间回退会导致产生重复id        long time = clock.millis();        Preconditions.checkState(lastTime <= time, "Clock is moving backwards, last time is %d milliseconds, current time is %d milliseconds", lastTime, time);        // 获取序列号        if (lastTime == time) {            if (0L == (sequence = ++sequence & SEQUENCE_MASK)) {                time = waitUntilNextTime(time);            }        } else {            sequence = 0;        }        // 设置最后时间戳        lastTime = time;        if (log.isDebugEnabled()) {            log.debug("{}-{}-{}", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(lastTime)), workerId, sequence);        }        // 生成编号        return ((time - SJDBC_EPOCH) << TIMESTAMP_LEFT_SHIFT_BITS) | (workerId << WORKER_ID_LEFT_SHIFT_BITS) | sequence;    }    //不停获得时间，直到大于最后时间    private long waitUntilNextTime(final long lastTime) {        long time = clock.millis();        while (time <= lastTime) {            time = clock.millis();        }        return time;    }}

  /**    * Shutdown the pool, closing all idle connections and aborting or closing    * active connections.    *    * @throws InterruptedException thrown if the thread is interrupted during shutdown    */   public synchronized void shutdown() throws InterruptedException {      try {         poolState = POOL_SHUTDOWN;         if (addConnectionExecutor == null) { // pool never started            return;         }         logPoolState("Before shutdown ");         if (houseKeeperTask != null) {            houseKeeperTask.cancel(false);            houseKeeperTask = null;         }         softEvictConnections();         addConnectionExecutor.shutdown();         addConnectionExecutor.awaitTermination(getLoginTimeout(), SECONDS);         destroyHouseKeepingExecutorService();         connectionBag.close();         final ExecutorService assassinExecutor = createThreadPoolExecutor(config.getMaximumPoolSize(), poolName + " connection assassinator",                                                                           config.getThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());         try {            final long start = currentTime();            do {               abortActiveConnections(assassinExecutor);               softEvictConnections();            } while (getTotalConnections() > 0 && elapsedMillis(start) < SECONDS.toMillis(10));         }         finally {            assassinExecutor.shutdown();            assassinExecutor.awaitTermination(10L, SECONDS);         }         shutdownNetworkTimeoutExecutor();         closeConnectionExecutor.shutdown();         closeConnectionExecutor.awaitTermination(10L, SECONDS);      }      finally {         logPoolState("After shutdown ");         unregisterMBeans();         metricsTracker.close();      }   }

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); //新添加的资源优先放入CopyOnWriteArrayList   // 当有等待资源的线程时，将资源交到某个等待线程后才返回（SynchronousQueue）   while (waiters.get() > 0 && !handoffQueue.offer(bagEntry)) {      yield();   }}public boolean remove(final T bagEntry) {   // 如果资源正在使用且无法进行状态切换，则返回失败   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); // 从CopyOnWriteArrayList中移出   if (!removed && !closed) {      LOGGER.warn("Attempt to remove an object from the bag that does not exist: {}", bagEntry);   }   return removed;}

public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException {   // 优先查看有没有可用的本地化的资源   final List<Object> list = threadList.get();   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;      }   }   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 long 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);      return null;   }   finally {      waiters.decrementAndGet();   }}public void requite(final T bagEntry) {   // 将状态转为未在使用   bagEntry.setState(STATE_NOT_IN_USE);   // 判断是否存在等待线程，若存在，则直接转手资源   for (int 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 {         yield();      }   }   // 否则，进行资源本地化   final List<Object> threadLocalList = threadList.get();   threadLocalList.add(weakThreadLocals ? new WeakReference<>(bagEntry) : bagEntry);}

private boolean useWeakThreadLocals(){   try {      // 人工指定是否使用弱引用，但是官方不推荐进行自主设置。      if (System.getProperty("com.dareway.concurrent.useWeakReferences") != null) {          return Boolean.getBoolean("com.dareway.concurrent.useWeakReferences");      }      // 默认通过判断初始化的ClassLoader是否是系统的ClassLoader来确定      return getClass().getClassLoader() != ClassLoader.getSystemClassLoader();   }   catch (SecurityException se) {      return true;   }}

/** * Entry used in the ConcurrentBag to track Connection instances. * * @author Brett Wooldridge */final class PoolEntry implements IConcurrentBagEntry {   private static final Logger LOGGER = LoggerFactory.getLogger(PoolEntry.class);   private static final AtomicIntegerFieldUpdater<PoolEntry> stateUpdater;   Connection connection;   long lastAccessed;   long lastBorrowed;   @SuppressWarnings("FieldCanBeLocal")   private volatile int state = 0;   private volatile boolean evict;   private volatile ScheduledFuture<?> endOfLife;   private final FastList<Statement> openStatements;   private final HikariPool hikariPool;   private final boolean isReadOnly;   private final boolean isAutoCommit;   static   {      stateUpdater = AtomicIntegerFieldUpdater.newUpdater(PoolEntry.class, "state");   }   PoolEntry(final Connection connection, final PoolBase pool, final boolean isReadOnly, final boolean isAutoCommit)   {      this.connection = connection;      this.hikariPool = (HikariPool) pool;      this.isReadOnly = isReadOnly;      this.isAutoCommit = isAutoCommit;      this.lastAccessed = currentTime();      this.openStatements = new FastList<>(Statement.class, 16);   }   /**    * Release this entry back to the pool.    *    * @param lastAccessed last access time-stamp    */   void recycle(final long lastAccessed) {      if (connection != null) {         this.lastAccessed = lastAccessed;         hikariPool.recycle(this);      }   }   /**    * Set the end of life {@link ScheduledFuture}.    *    * @param endOfLife this PoolEntry/Connection's end of life {@link ScheduledFuture}    */   void setFutureEol(final ScheduledFuture<?> endOfLife) {      this.endOfLife = endOfLife;   }   Connection createProxyConnection(final ProxyLeakTask leakTask, final long now) {      return ProxyFactory.getProxyConnection(this, connection, openStatements, leakTask, now, isReadOnly, isAutoCommit);   }   void resetConnectionState(final ProxyConnection proxyConnection, final int dirtyBits) throws SQLException {      hikariPool.resetConnectionState(connection, proxyConnection, dirtyBits);   }   String getPoolName() {      return hikariPool.toString();   }   boolean isMarkedEvicted() {      return evict;   }   void markEvicted() {      this.evict = true;   }   void evict(final String closureReason) {      hikariPool.closeConnection(this, closureReason);   }   /** Returns millis since lastBorrowed */   long getMillisSinceBorrowed() {      return elapsedMillis(lastBorrowed);   }   /** {@inheritDoc} */   @Override   public String toString() {      final long now = currentTime();      return connection         + ", accessed " + elapsedDisplayString(lastAccessed, now) + " ago, "         + stateToString();   }   // ***********************************************************************   //                      IConcurrentBagEntry methods   // ***********************************************************************   /** {@inheritDoc} */   @Override   public int getState() {      return stateUpdater.get(this);   }   /** {@inheritDoc} */   @Override   public boolean compareAndSet(int expect, int update) {      return stateUpdater.compareAndSet(this, expect, update);   }   /** {@inheritDoc} */   @Override   public void setState(int update) {      stateUpdater.set(this, update);   }   Connection close() {      ScheduledFuture<?> eol = endOfLife;      if (eol != null && !eol.isDone() && !eol.cancel(false)) {         LOGGER.warn("{} - maxLifeTime expiration task cancellation unexpectedly returned false for connection {}", getPoolName(), connection);      }      Connection con = connection;      connection = null;      endOfLife = null;      return con;   }   private String stateToString() {      switch (state) {      case STATE_IN_USE:         return "IN_USE";      case STATE_NOT_IN_USE:         return "NOT_IN_USE";      case STATE_REMOVED:         return "REMOVED";      case STATE_RESERVED:         return "RESERVED";      default:         return "Invalid";      }   }}