雨羊的博客

// DruidPooledConnection 类的 close 方法
public void close() throws SQLException {
    //检查，因为该连接对象是抛出去给别的业务线程使用，也就是说并不受连接池本身管控，所以很可能存在多线程同时 close 的操作，因此这里需要做一层检查，包括下方的 syncClose 里的检查也是一个意思
    if (this.disable) {
        return;
    }

    // 拿到对应的 holder 对象（之前说过，这个对象才是最后放进连接池的对象）
    DruidConnectionHolder holder = this.holder;
    if (holder == null) {
        if (dupCloseLogEnable) {
            LOG.error("dup close");
        }
        return;
    }

    //拿到对应的连接池对象
    DruidAbstractDataSource dataSource = holder.getDataSource();
    boolean isSameThread = this.getOwnerThread() == Thread.currentThread();

    // 关闭该连接与获取该连接的线程并非同一个的时候，则触发下面的 syncClose
    if (!isSameThread) {
        dataSource.setAsyncCloseConnectionEnable(true);
    }

    if (dataSource.isAsyncCloseConnectionEnable()) {
        // 参考上面的解释，该方法详情在下方
        syncClose();
        return;
    }

    if (!CLOSING_UPDATER.compareAndSet(this, 0, 1)) {
        return;
    }

    try {
        // 一些事件监听器的触发，忽略
        for (ConnectionEventListener listener : holder.getConnectionEventListeners()) {
            listener.connectionClosed(new ConnectionEvent(this));
        }

        // 责任链的执行，参考流程 1.1 与代码段 1-2，运行方式是一样的，找到映射方法，整个触发一遍责任链上的 filters
        List<Filter> filters = dataSource.getProxyFilters();
        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(dataSource);
            filterChain.dataSource_recycle(this);
        } else {
            // 触发目标方法 recycle
            recycle();
        }
    } finally {
        CLOSING_UPDATER.set(this, 0);
    }

    // 标记该连接已失效，无法再次提供服务
    this.disable = true;
}

// 上面逻辑走 syncClose 的情况，该方法与上面大体相同，但由于不是同一个线程做的操作，所以这里需要锁控制
public void syncClose() throws SQLException {
    lock.lock();
    try {
        if (this.disable || CLOSING_UPDATER.get(this) != 0) {
            return;
        }

        // 同样的，拿到需要归还的 holder 对象
        DruidConnectionHolder holder = this.holder;
        if (holder == null) {
            if (dupCloseLogEnable) {
                LOG.error("dup close");
            }
            return;
        }

        if (!CLOSING_UPDATER.compareAndSet(this, 0, 1)) {
            return;
        }

        // 同样是一些事件监听器的触发，忽略
        for (ConnectionEventListener listener : holder.getConnectionEventListeners()) {
            listener.connectionClosed(new ConnectionEvent(this));
        }

        // 同样的责任链的执行，参考上面的解释
        DruidAbstractDataSource dataSource = holder.getDataSource();
        List<Filter> filters = dataSource.getProxyFilters();
        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(dataSource);
            filterChain.dataSource_recycle(this);
        } else {
            // 触发目标方法 recycle，方法详情在下方
            recycle();
        }

        // 标记该连接已失效，无法再次提供服务
        this.disable = true;
    } finally {
        CLOSING_UPDATER.set(this, 0);
        // 解锁
        lock.unlock();
    }
}

// DruidPooledConnection 类的 recycle 方法，由上面俩方法直接触发
public void recycle() throws SQLException {
    if (this.disable) {
        return;
    }

    // 拿到真正需要归还的连接对象
    DruidConnectionHolder holder = this.holder;
    if (holder == null) {
        if (dupCloseLogEnable) {
            LOG.error("dup close");
        }
        return;
    }

    // 如果期间已经被流程 4.2 处理过了（abandoned==true），则不触发下方逻辑
    if (!this.abandoned) {
        DruidAbstractDataSource dataSource = holder.getDataSource();
        // 真正触发连接池的回收方法，方法详情在下方
        dataSource.recycle(this);
    }

    // 连接对象一旦被回收处理，则会把所有与连接相关的属性置空（不持有），closed 标记为 true
    this.holder = null;
    conn = null;
    transactionInfo = null;
    closed = true;
}

// DruidDataSource 类里的 recycle 方法，真正回收连接的方法，由上面 DruidPooledConnection 类的 recycle 触发
protected void recycle(DruidPooledConnection pooledConnection) throws SQLException {
    final DruidConnectionHolder holder = pooledConnection.holder;

    if (holder == null) {
        LOG.warn("connectionHolder is null");
        return;
    }

    if (logDifferentThread //
        && (!isAsyncCloseConnectionEnable()) //
        && pooledConnection.ownerThread != Thread.currentThread()//
    ) {
        LOG.warn("get/close not same thread");
    }

    // 拿到真正的驱动连接对象
    final Connection physicalConnection = holder.conn;

    // 如果 traceEnable 为 true（满足该属性为 true，必须要 removeAbandoned 设置为 true，这样在主流程1那里才会被放进 activeConnections，才会置为 true），流程4.2处理过后，会把该属性重新置为 false，其他情况均为 true
    if (pooledConnection.traceEnable) {
        Object oldInfo = null;
        activeConnectionLock.lock();
        try {
            // 双重检查
            if (pooledConnection.traceEnable) {
                // 从 activeConnections 移除，防止流程 4.2 的重复检查
                oldInfo = activeConnections.remove(pooledConnection);
                // 置为 false
                pooledConnection.traceEnable = false;
            }
        } finally {
            activeConnectionLock.unlock();
        }
        if (oldInfo == null) {
            if (LOG.isWarnEnabled()) {
                LOG.warn("remove abandonded failed. activeConnections.size " + activeConnections.size());
            }
        }
    }

    final boolean isAutoCommit = holder.underlyingAutoCommit;
    final boolean isReadOnly = holder.underlyingReadOnly;
    final boolean testOnReturn = this.testOnReturn;

    try {
        // 如果在归还至连接池时发现此连接对象还有未处理完的事务，则直接回滚
        // check need to rollback?
        if ((!isAutoCommit) && (!isReadOnly)) {
            pooledConnection.rollback();
        }

        // reset holder, restore default settings, clear warnings
        boolean isSameThread = pooledConnection.ownerThread == Thread.currentThread();
        // 同样判断线程，为了保证安全性
        if (!isSameThread) {
            final ReentrantLock lock = pooledConnection.lock;
            lock.lock();
            try {
                //连接被借出去后，可能被业务方改动了一些属性（典型的比如 autoCommit），现在利用 reset 方法还原为默认值
                holder.reset();
            } finally {
                lock.unlock();
            }
        } else {
            // 同上，这里认为获取和关闭连接的是同一个线程，不存在线程安全问题，因此不用去竞争锁
            holder.reset();
        }

        // 连接已被抛弃，则不作任何处理（不再归还）
        if (holder.discard) {
            return;
        }

        //忽略
        if (phyMaxUseCount > 0 && holder.useCount >= phyMaxUseCount) {
            discardConnection(holder);
            return;
        }

        // 如果驱动连接本身被人为关闭了，除一些监控值之外，也不做处理
        if (physicalConnection.isClosed()) {
            lock.lock();
            try {
                if (holder.active) {
                    activeCount--;
                    holder.active = false;
                }
                closeCount++;
            } finally {
                lock.unlock();
            }
            return;
        }

        // 参考 testOnBorrow，这里 testOnReturn 就是指每次回收连接都要做连接可用性检查，同样官方不建议开启，影响性能，缺省值也是不开启的
        if (testOnReturn) {
            // 流程忽略
            boolean validate = testConnectionInternal(holder, physicalConnection);
            if (!validate) {
                JdbcUtils.close(physicalConnection);

                destroyCountUpdater.incrementAndGet(this);

                lock.lock();
                try {
                    if (holder.active) {
                        activeCount--;
                        holder.active = false;
                    }
                    closeCount++;
                } finally {
                    lock.unlock();
                }
                return;
            }
        }
        if (holder.initSchema != null) {
            holder.conn.setSchema(holder.initSchema);
            holder.initSchema = null;
        }

        // 中途发现连接又被置为不可用，则直接触发抛弃方法，参考流程 1.4 和代码段 1-5
        if (!enable) {
            discardConnection(holder);
            return;
        }

        boolean result;
        final long currentTimeMillis = System.currentTimeMillis();

        if (phyTimeoutMillis > 0) {
            long phyConnectTimeMillis = currentTimeMillis - holder.connectTimeMillis;
            if (phyConnectTimeMillis > phyTimeoutMillis) {
                discardConnection(holder);
                return;
            }
        }

        lock.lock();
        try {
            if (holder.active) {
                activeCount--;
                holder.active = false;
            }
            closeCount++;

            // 最终放入池子，方法详情在下方
            result = putLast(holder, currentTimeMillis);
            recycleCount++;
        } finally {
            lock.unlock();
        }

        // 如果加不进去，则直接关闭驱动连接，然后不处理（此时 holder 已经失去强引用，不久便会被回收）
        if (!result) {
            JdbcUtils.close(holder.conn);
            LOG.info("connection recyle failed.");
        }
    } catch (Throwable e) {
        holder.clearStatementCache();

        if (!holder.discard) {
            discardConnection(holder);
            holder.discard = true;
        }

        LOG.error("recyle error", e);
        recycleErrorCountUpdater.incrementAndGet(this);
    }
}

// DruidDataSource 类里的 putLast 方法，由上方的 recycle 方法触发
boolean putLast(DruidConnectionHolder e, long lastActiveTimeMillis) {
    // 池子已满，不加
    if (poolingCount >= maxActive || e.discard || this.closed) {
        return false;
    }

    // 刷新上次活跃时间，该时间很重要，直接影响连接检查的触发
    e.lastActiveTimeMillis = lastActiveTimeMillis;
    // 放进连接池数组尾部
    connections[poolingCount] = e;
    incrementPoolingCount();

    if (poolingCount > poolingPeak) {
        poolingPeak = poolingCount;
        poolingPeakTime = lastActiveTimeMillis;
    }

    // 因为成功回收了一个连接，那就唤起一次所有因为获取不到连接而被阻塞的业务线程吧~（参考流程1.2）
    notEmpty.signal();
    notEmptySignalCount++;

    return true;
}
如果觉得文章对你有用，请随意赞赏
druid 连接池
druid 回收连接
https://blog.rainsheep.cn/archives/1646382661977
作者

Rainsheep
发布于

2022-03-04
更新于

2023-12-28
许可协议

CC BY 4.0