Druid 连接池介绍

DruidDataSource共享锁

public abstract class DruidAbstractDataSource extends WrapperAdapter implements DruidAbstractDataSourceMBean, DataSource, DataSourceProxy, Serializable {
    // 可重入锁 lock
    protected ReentrantLock                            lock;
    // 非空条件变量
    protected Condition                                notEmpty;
    // 空条件变量
    protected Condition                                empty;

    public DruidAbstractDataSource(boolean lockFair){
        lock = new ReentrantLock(lockFair);
        notEmpty = lock.newCondition();
        empty = lock.newCondition();
    }
}

public class DruidDataSource extends DruidAbstractDataSource implements DruidDataSourceMBean, ManagedDataSource, Referenceable, Closeable, Cloneable, ConnectionPoolDataSource, MBeanRegistration {

    private volatile DruidConnectionHolder[] connections;
    private DruidConnectionHolder[]          evictConnections;
    private DruidConnectionHolder[]          keepAliveConnections;

    private CreateConnectionThread           createConnectionThread;
    private DestroyConnectionThread          destroyConnectionThread;
    private LogStatsThread                   logStatsThread;
}

• DruidDataSource的三个变量通过构造函数初始化，可以指定采用公平锁或者非公平锁。
• 生产者和消费者的任何操作都需要获得lock。生产者根据empty条件变量await，当连接池中连接被消耗完后触发empty的通知的时候，阻塞在empty上的生产者开始创建连接。创建完成之后生产者发送notEmpty的sigal信号，触发在notEmpty上的消费者来获取连接进行消费。这是Druid连接池的基本原理。
• DruidDataSource连接池的缓冲区位于DruidDataSource中的DruidConnectionHolder[]数组中。

DruidDataSource连接池

• CreateConnectionThread是连接池中的创建连接的线程，当连接池中的连接不足的时候负责创建连接。
• DestroyConnectionThread是连接池中的销毁连接线程，当连接池中出现空闲连接超过配置的空闲连接数，或者出现一些不健康的连接，那么线程池将会通过DestroyConnectionThread线程将连接回收。

DruidDataSource 核心线程

    public class CreateConnectionThread extends Thread {

        public CreateConnectionThread(String name){
            super(name);
            this.setDaemon(true);
        }

        public void run() {

            initedLatch.countDown();

            long lastDiscardCount = 0;
            int errorCount = 0;
            for (;;) {
                // addLast
                try {
                    lock.lockInterruptibly();
                } catch (InterruptedException e2) {
                    break;
                }

                long discardCount = DruidDataSource.this.discardCount;
                boolean discardChanged = discardCount - lastDiscardCount > 0;
                lastDiscardCount = discardCount;

                try {
                    boolean emptyWait = true;

                    if (createError != null
                            && poolingCount == 0
                            && !discardChanged) {
                        emptyWait = false;
                    }

                    if (emptyWait
                            && asyncInit && createCount < initialSize) {
                        emptyWait = false;
                    }
                    // 判断是否需要等待连接池为空的信号量 empty.wait()
                    if (emptyWait) {
                        // 必须存在线程等待，才创建连接
                        if (poolingCount >= notEmptyWaitThreadCount //
                                && (!(keepAlive && activeCount + poolingCount < minIdle))
                                && !isFailContinuous()
                        ) {
                            empty.await();
                        }

                        // 防止创建超过maxActive数量的连接
                        if (activeCount + poolingCount >= maxActive) {
                            empty.await();
                            continue;
                        }
                    }

                } catch (InterruptedException e) {
                    break;
                } finally {
                    lock.unlock();
                }

                // 负责创建连接
                PhysicalConnectionInfo connection = null;
                try {
                    connection = createPhysicalConnection();
                } catch (Error e) {
                    setFailContinuous(true);
                    break;
                }

                // 添加新建的连接
                boolean result = put(connection);

                errorCount = 0; // reset errorCount
            }
        }
    }

• CreateConnectionThread负责创建连接放到连接池当中，当连接池的连接数足够的时候通过empty.await()进行阻塞等待，否则创建连接通过 put 放到连接池。

    public class DestroyConnectionThread extends Thread {

        public DestroyConnectionThread(String name){
            super(name);
            this.setDaemon(true);
        }

        public void run() {
            initedLatch.countDown();

            for (;;) {
                // 从前面开始删除
                try {
                    if (closed) {
                        break;
                    }
                    // 等待连接驱除等待时间
                    if (timeBetweenEvictionRunsMillis > 0) {
                        Thread.sleep(timeBetweenEvictionRunsMillis);
                    } else {
                        Thread.sleep(1000); //
                    }

                    if (Thread.interrupted()) {
                        break;
                    }

                    destroyTask.run();
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }

    public class DestroyTask implements Runnable {
        public DestroyTask() {
        }

        @Override
        public void run() {
            // 缩减连接池
            shrink(true, keepAlive);

            if (isRemoveAbandoned()) {
                removeAbandoned();
            }
        }

    }

• DestroyConnectionThread负责定期调整连接池里面的连接。
• 判断连接池的连接空闲数是否大于minIdle，如果是则关闭多余的连接，反之则新增连接，具体可以参见：Druid配置参数详解-minIdle
• 回收连接池泄露的连接，具体可以参见：Druid配置参数详解-removeAbandoned

Druid 监控参数

Druid 配置参数