druid解析flink sql druid源码解析

概述

Druid是阿里巴巴开源的一个数据库连接池 源码地址。下面简单分析一下连接池是怎么实现的

怎么开始阅读

如果使用过Druid连接池的都只要在Spring配置中配置jdbc的时候配置Driver是使用的DruidDataSource。因此，在读源码的时候也可以从这个类入口。

Datasouce

什么是Datasouce呢，其实就是用于管理数据库连接的工厂类。接口就2个方法

public interface DataSource  extends CommonDataSource, Wrapper {
  Connection getConnection() throws SQLException;
  Connection getConnection(String username, String password)
    throws SQLException;
}

DruidDataSource

DruidDataSource就是实现了这个接口，利用池化思想来管理数据库连接。池化的思想我理解的主要有2个目的：

• 一个目的是可以重复利用一些资源，特别是那些创建和销毁的开销都比较大的资源
• 一个是可以控制资源的数量，防止大规模的创建导致系统问题 因此，DruidDataSource的关键就是在调用getConnection() 的时候从连接池中获取正真的数据库连接，并且在关闭连接的时候并不是真正的关闭物理连接，而是把连接重新放到连接池中。

创建连接池

public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {
        init();

        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(this);
            return filterChain.dataSource_connect(this, maxWaitMillis);
        } else {
            return getConnectionDirect(maxWaitMillis);
        }
    }

init()就是初始化连接池，其中核心代码：

public void init() throws SQLException {
    ...
    //line : 845
    for (int i = 0, size = getInitialSize(); i < size; ++i) {
        //看名字就是知道是保存物理连接的类。并且在这里会实际创建物理连接（JDBC的Connection)，
        //ps.准确的是说是ConnectionProxyImpl类，这个类是实现监控的关键，后面会再写一篇文章介绍
        PhysicalConnectionInfo pyConnectInfo = createPhysicalConnection();
        //这种构造的时候传入 this和另外一个对象一般情况都是包装类，
        //这样在DruidConnectionHolder就可以获取DruidDataSource的一些状态字段和成员对象（连接归还的时候就会用到）
        DruidConnectionHolder holder = new DruidConnectionHolder(this, pyConnectInfo);
        //connections 保存连接的数组
        connections[poolingCount] = holder;
        //方法就一行代码:poolingCount++; 从属性的名字推断就是对连接池中的连接计数
        //初始化完成首poolingCount的值就等于初始化连接的数量
        incrementPoolingCount();
    }
    ...

初始化完成后就看怎么获取连接，回到上面getConnection()的方法中，直接看getConnectionDirect()方法吧。（我相信有过滤器的创建连接最终肯定还是调用这个方法,只不过这里会用到责任链模式来处理过滤器，可以参考之前的文章介绍责任链实现方式)。下面看看

public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {
    int notFullTimeoutRetryCnt = 0;
    for (;;) {
        DruidPooledConnection poolableConnection;
        try {
            poolableConnection = getConnectionInternal(maxWaitMillis);
        } catch (GetConnectionTimeoutException ex) {
           ...
        }

        //这里都是做一些配置的校验，比如配置了testOnBorrow，那么在这里会对连接进行测试
        ...

        return poolableConnection;
    }
}

private DruidPooledConnection getConnectionInternal(long maxWait) throws SQLException {
    DruidConnectionHolder holder;
    ...
    if (maxWait > 0) {
        holder = pollLast(nanos);
    } else {
        holder = takeLast();
    }

    if (holder != null) {
        activeCount++;
        if (activeCount > activePeak) {
            activePeak = activeCount;
            activePeakTime = System.currentTimeMillis();
        }
    }
   ...

    DruidPooledConnection poolalbeConnection = new DruidPooledConnection(holder);
    return poolalbeConnection;
}

这里看到DruidConnectionHolder，也就是再初始化的时候生成的包含了物理连接的保证类，那么pollLast(nanos)肯定就是有超时时间的获取，takeLast()肯定就是无超时时间的获取，那么直接看takeLast()

DruidConnectionHolder takeLast() throws InterruptedException, SQLException {
        try {
            while (poolingCount == 0) {
                emptySignal(); // send signal to CreateThread create connection

                ...
                notEmpty.await(); // signal by recycle or creator
                ...
            }
        } catch (InterruptedException ie) {
            ...
        }
        //poolingCount--
        decrementPoolingCount();
        DruidConnectionHolder last = connections[poolingCount];
        connections[poolingCount] = null;

        return last;
    }

 

这里就是如何从连接池获取连接的核心代码了，这里poolingCount为0的情况下就会发送empty信号（回想一下自己刚开始写生产者消费者的代码吧，是不是用的到了一个empty和full来控制消费队列为空和满的情况），这里也是这样的，当poolingCount==0的时候就表示没有可用的连接。

• 如果达到最大连接数，阻塞
• 如果没有达到，创建新的连接，这里创建新的连接是通过一个线程去执行的，详情参考CreateConnectionTask。

当然如果当poolingCount不为0的时候，那么直接从连接数组中获取下表为当poolingCount-1的连接返回就可以啦。

连接关闭

看了上面如何从连接池中获取连接，那么很自然都可以知道如何把连接放回连接池中，肯定就是 connections[poolingCount] = 待返回的连接，然后poolingCount+1。

/**
 * 回收连接
 * DruidDataSouce.java
 */
protected void recycle(DruidPooledConnection pooledConnection) throws SQLException {
    final DruidConnectionHolder holder = pooledConnection.getConnectionHolder();

    result = putLast(holder, lastActiveTimeMillis);
           ....
}

boolean putLast(DruidConnectionHolder e, long lastActiveTimeMillis) {
    if (poolingCount >= maxActive) {
        return false;
    }

    e.setLastActiveTimeMillis(lastActiveTimeMillis);
    connections[poolingCount] = e;
    incrementPoolingCount();

    notEmpty.signal();
    return true;
}

果然和我们想的一样，不过这里还有一个很重要的一部，调用notEmpty.signal();

小结

从复杂逻辑中把连接池的相关逻辑抽取出来，其实就很简单，类似于一个生产者消费者模型。希望的这篇文章对你有帮助。