tomcat-jdbc源码学习

比眉伴天荒 2024-02-19 16:52 21阅读 0赞

##### DataSource工厂 ： DataSourceFactory #####

1.  DataSource工厂根据配置创建数数据源
2.  parsePoolProperties解析properties创建PoolConfiguration poolProperties = new PoolProperties();
3.  创建数据源org.apache.tomcat.jdbc.pool.DataSource
4.  创建连接池DataSourceProxy.createPool（ConnectionPool）

##### 初始化连接池 #####

1.  初始化线程池ConnectionPool.init
2.  校验线程池参数checkPoolConfiguration
3.  创建busy队列LinkedBlockingQueue
4.  创建idle队列，如果指定为公平的则为FairBlockingQueue，否则LinkedBlockingQueue
5.  初始化线程池清理线程池任务initializePoolCleaner

public void initializePoolCleaner(PoolConfiguration properties) {
        //if the evictor thread is supposed to run, start it now
        if (properties.isPoolSweeperEnabled()) {
            poolCleaner = new PoolCleaner(this, properties.getTimeBetweenEvictionRunsMillis());
            poolCleaner.start();
        } //end if
    }
    PoolCleaner(ConnectionPool pool, long sleepTime) {
        this.pool = new WeakReference<>(pool);
        this.sleepTime = sleepTime;
        if (sleepTime <= 0) {
            log.warn("Database connection pool evicter thread interval is set to 0, defaulting to 30 seconds");
            this.sleepTime = 1000 * 30;
        } else if (sleepTime < 1000) {
            log.warn("Database connection pool evicter thread interval is set to lower than 1 second.");
        }
    }

##### 线程池清理任务PoolCleaner #####

1.  注册registerCleaner定时任务
2.  线程池清理的pool实例属性是对线程池的一个弱引用
3.  弱引用get线程池对象，如果为空，注销定时任务
4.  如果pool没有close则进行一系列检查

*  ##### 配置启用移除abandon连接或者suspectTimeout大于0，检查abandon连接 #####
 *  ##### checkAbandoned检查abandon连接 #####

1.  遍历busy队列
2.  如果idle队列包含该连接或者已经释放，跳过该连接
3.  是否应该abandon并且（当前时间-上次连接池"touched"连接时间）已经超过abandonTimeout配置

protected boolean shouldAbandon() {
        if (!poolProperties.isRemoveAbandoned()) return false;
        if (poolProperties.getAbandonWhenPercentageFull()==0) return true;
        float used = busy.size();
        float max  = poolProperties.getMaxActive();
        float perc = poolProperties.getAbandonWhenPercentageFull();
        return (used/max*100f)>=perc;
    }

1.  是则busy中移除该连接并且标识设置null标识setToNull，执行abandon，释放链接，发送通知至jmx
2.  suspectTimeout不为0并且（当前时间-上次连接池"touched"(上次borrowConnection的时间)连接时间）已经超过配置，则suspect
3.  如果已经处于被怀疑标识返回，否则打标识

*  ##### 线程池idle队列size大于minIdle #####
 *  ##### checkIdle检查idle连接 #####

1.  遍历idle队列
2.  busy队列包含该连接，跳过该链接校验
3.  shouldReleaseIdle是否应该释放连接

protected boolean shouldReleaseIdle(long now, PooledConnection con, long time) {
        if (con.getConnectionVersion() < getPoolVersion()) return true;
        else return (con.getReleaseTime()>0) && ((now - time) > con.getReleaseTime()) && (getSize()>getPoolProperties().getMinIdle());
    }

1.  版本小于当前版本，释放链接
2.  getMinEvictableIdleTimeMillis大于0并且当前时间-上次连接池"touched"连接时间超过minEvictableIdleTimeMillis配置并且当前连接池size大于minIdle，释放链接

*  ##### testAllIdle测试所有空闲连接 #####

1.  遍历idle队列
2.  如果busy包含该连接则跳过该链接
3.  校验validate
4.  如果getValidator自定义校验不为空使用自定义校验规则校验
5.  否则使用initsql校验，没有initsql则使用getValidationQuery校验sql进行校验，如果没有校验sql则使用jdbc驱动校验方式校验（超时时间为validationQueryTimeout）
6.  存在校验sql，则执行校验sql校验，校验成功更新校验时间戳，
7.  校验失败返回false并释放连接

*  ##### 如果启用jmx则create JMX MBean #####
 *  ##### 如果配置了JdbcInterceptor拦截器，则通知连接池已启动 #####
 *  ##### 初始化线程池 #####

1.  按照initialSize初始化大小配置初始化
2.  borrowConnection获取连接
3.  拉取idle队列，如果不为空，连接该连接并返回borrowConnection，borrowedCount递增

>  *  3.1 判断是否需要强制重连forceReconnect=shouldForceReconnect用户名密码是否变更，或者isMaxAgeExpired上次连接时间距离当前时间是否超过maxAge配置
>  *  3.2 如果连接被discarded并且没有初始化，forceReconnect置为true
>  *  3.3 如果不需要强制重连，连接没有被discarded并且validateOnBorrow成功（校验逻辑执行成功），设置timestamp时间戳（即一次touched）并将怀疑超时对象标识设置为false，将连接offer提交至busy（失败则打印debug日志），返回连接
>  *  3.4 如果走到这步说明要么连接已经有了另外一个实体，被废弃或者校验失败
>  *  3.5 重连并递增reconnectedCount
>  *  3.6 如果testOnConnect或者initsql不为空，则validateInit，否则如果开启了testOnBorrow则validateBorrow
>  *  3.7 如果校验通过添加至busy，添加失败打印debug日志：“Connection doesn’t fit into busy array, connection will not be traceable.”
>  *  3.8 如果校验失败，打印：“Failed to validate a newly established connection.”

1.  连接池size小于maxActive配置
2.  如果size递增后大于maxActive，则减1恢复size后继续，否则创建连接createConnection

//if we get here, see if we need to create one
    //this is not 100% accurate since it doesn't use a shared
    //atomic variable - a connection can become idle while we are creating
    //a new connection
    if (size.get() < getPoolProperties().getMaxActive()) {
        //atomic duplicate check
        if (size.addAndGet(1) > getPoolProperties().getMaxActive()) {
            //if we got here, two threads passed through the first if
            size.decrementAndGet();
        } else {
            //create a connection, we're below the limit
            return createConnection(now, con, username, password);
        }
    } //end if
    
    //calculate wait time for this iteration
    long maxWait = wait;

>  *  5.1 创建PooledConnection，保存用户密码
>  *  5.2 PooledConnection.connect连接，如果已经release抛出异常：“A connection once released, can’t be reestablished.”
>  *  5.3 如果connection不为空，先断开连接再连接，如果存在DataSource调用connectUsingDataSource，否则调用jdbc驱动连接，返回连接

1.  走到此处说明没有idle并且连接数大于等于maxActive，则等待空闲，timetowait=获取maxWait配置计算等待超时时间（maxWait-前两步花费的时间），等待次数递增waitcount
2.  idle拉取超时时间为timetowait（即：每次获取空闲连接失败并且创建失败，则等待空闲队列空闲的时间为maxWait减去之前两步花去的时间，3步总共花费maxWait毫秒数）
3.  超时报错抛出异常"Pool wait interrupted…"
4.  如果maxWait=0并且队列拉取的con为空，抛出异常：“NoWait: Pool empty. Unable to fetch a connection, none available\[”+busy.size()+" in use\]."
5.  如果maxWait不为空并且队列拉取的con为空，并且当前时间-now（开始循环逻辑的时间）大于等于maxWait，则抛出异常

throw new PoolExhaustedException("[" + Thread.currentThread().getName()+"] " +
                        "Timeout: Pool empty. Unable to fetch a connection in " + (maxWait / 1000) +
                        " seconds, none available[size:"+size.get() +"; busy:"+busy.size()+"; idle:"+idle.size()+"; lastwait:"+timetowait+"].");

1.  否则代表未超时继续循环
2.  成功则返回PooledConnection，完成初始化的borrowConnection流程（即初始化连接数的PooledConnection创建）
3.  创建失败则抛出异常：“Unable to create initial connections of pool.”
4.  遍历初始化连接池调用returnConnection

finally {
        //return the members as idle to the pool
        for (int i = 0; i < initialPool.length; i++) {
            if (initialPool[i] != null) {
                try {this.returnConnection(initialPool[i]);}catch(Exception x){/*NOOP*/}
            } //end if
        } //for
    } //catch

*  ##### returnConnection：return the members as idle to the pool，Returns a connection to the pool #####

1.  如果已经close则释放连接
2.  如果入参PooledConnection con不为空，returnedCount递增
3.  如果连接已经被标注为被怀疑超时的对象，日志开启的情况下则打印日志
4.  busy队列移除该连接成功，如果shouldClose则释放并打印debug日志：“Connection \[”+con+"\] will be closed and not returned to the pool."
5.  如果shouldClose为false

protected boolean shouldClose(PooledConnection con, int action) {
        if (con.getConnectionVersion() < getPoolVersion()) return true;
        if (con.isDiscarded()) return true;
        if (isClosed()) return true;
        if (!con.validate(action)) return true;
        if (!terminateTransaction(con)) return true;
        if (con.isMaxAgeExpired()) return true;
        else return false;
    }

1.  如果idle大于等于最大值并且没有开启清扫任务，不提交连接至idle，释放连接，否则，提交连接至idle，如果offer提交失败，释放连接；打印debug日志

if (((idle.size()>=poolProperties.getMaxIdle()) && !poolProperties.isPoolSweeperEnabled()) || (!idle.offer(con))) {
        if (log.isDebugEnabled()) {
            log.debug("Connection ["+con+"] will be closed and not returned to the pool, idle["+idle.size()+"]>=maxIdle["+poolProperties.getMaxIdle()+"] idle.offer failed.");
        }
        release(con);
    }

1.  isPooledSweeperEnabled是否开启了清扫任务并且配置合理

>  *  7.1 evict驱逐任务周期不为0，并且开启了removeAbandon移除狂热对象并且移除狂热对象超时配置大于0，直接返回true。

public boolean isPoolSweeperEnabled() {
        boolean timer = getTimeBetweenEvictionRunsMillis()>0;
        boolean result = timer && (isRemoveAbandoned() && getRemoveAbandonedTimeout()>0);
        result = result || (timer && getSuspectTimeout()>0);
        result = result || (timer && isTestWhileIdle() && getValidationQuery()!=null);
        result = result || (timer && getMinEvictableIdleTimeMillis()>0);
        return result;
    }

1.  busy队列移除该连接失败，释放连接，打印debug日志：“Connection \[”+con+"\] will be closed and not returned to the pool, busy.remove failed."
2.  初始化完成

*  ##### DataSource.ConnectionPool.getConnection #####

1.  从连接池中获取连接

public Connection getConnection() throws SQLException {
        //check out a connection
        PooledConnection con = borrowConnection(-1,null,null);
        return setupConnection(con);
    }

*  ##### FairBlockingQueue（idle队列的可配置实现） #####
 *  offer超时方法未实现，实际调用的offer无超时方法
 *  offer提交
 *  如果waiters大于0说明已经开始排队，则当前提交的对象附加到CountDownLatch上并且countDown唤醒异步拉取的任务

public boolean offer(E e) {
        //during the offer, we will grab the main lock
        final ReentrantLock lock = this.lock;
        lock.lock();
        ExchangeCountDownLatch<E> c = null;
        try {
            //check to see if threads are waiting for an object
            if (waiters.size() > 0) {
                //if threads are waiting grab the latch for that thread
                c = waiters.poll();
                //give the object to the thread instead of adding it to the pool
                c.setItem(e);
                if (isLinux) c.countDown();
            } else {
                //we always add first, so that the most recently used object will be given out
                items.addFirst(e);
            }
        } finally {
            lock.unlock();
        }
        //if we exchanged an object with another thread, wake it up.
        if (!isLinux && c!=null) c.countDown();
        //we have an unbounded queue, so always return true
        return true;
    }

*  否则添加至items头部addFirst
 *  poll同步拉取，直接调用items的linkedlist的poll方法

public E poll() {
        ReentrantLock lock = this.lock;
        lock.lock();
    
        Object var2;
        try {
            var2 = this.items.poll();
        } finally {
            lock.unlock();
        }
    
        return var2;
    }

*  poll异步拉取
 *  如果items拉取为空，则将拉取任务以CountDownLatch方式提交至waiters，返回CountDownLatch的Future

public Future<E> pollAsync() {
        Future<E> result = null;
        final ReentrantLock lock = this.lock;
        //grab the global lock
        lock.lock();
        try {
            //check to see if we have objects in the queue
            E item = items.poll();
            if (item==null) {
                //queue is empty, add ourselves as waiters
                ExchangeCountDownLatch<E> c = new ExchangeCountDownLatch<>(1);
                waiters.addLast(c);
                //return a future that will wait for the object
                result = new ItemFuture<>(c);
            } else {
                //return a future with the item
                result = new ItemFuture<>(item);
            }
        } finally {
            lock.unlock();
        }
        return result;
    }

*  poll超时拉取
 *  如果items拉取为空，并且超时配置大于0
 *  将拉取任务以CountDownLatch方式提交至waiters，CountDownLatch等待设置的超时时间，如果超时waiters移除该次任务并返回null，否则获取正常的CountDownLatch附加的元素返回

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E result = null;
        final ReentrantLock lock = this.lock;
        //acquire the global lock until we know what to do
        lock.lock();
        try {
            //check to see if we have objects
            result = items.poll();
            if (result==null && timeout>0) {
                //the queue is empty we will wait for an object
                ExchangeCountDownLatch<E> c = new ExchangeCountDownLatch<>(1);
                //add to the bottom of the wait list
                waiters.addLast(c);
                //unlock the global lock
                lock.unlock();
                boolean didtimeout = true;
                InterruptedException interruptedException = null;
                try {
                    //wait for the specified timeout
                    didtimeout = !c.await(timeout, unit);
                } catch (InterruptedException ix) {
                    interruptedException = ix;
                }
                if (didtimeout) {
                    //if we timed out, or got interrupted
                    // remove ourselves from the waitlist
                    lock.lock();
                    try {
                        waiters.remove(c);
                    } finally {
                        lock.unlock();
                    }
                }
                //return the item we received, can be null if we timed out
                result = c.getItem();
                if (null!=interruptedException) {
                    //we got interrupted
                    if ( null!=result) {
                        //we got a result - clear the interrupt status
                        //don't propagate cause we have removed a connection from pool
                        Thread.interrupted();
                    } else {
                        throw interruptedException;
                    }
                }
            } else {
                //we have an object, release
                lock.unlock();
            }
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
        return result;
    }