现在的位置: 首页 > 综合 > 正文

XAPool连接池源码分析

2013年11月05日 ⁄ 综合 ⁄ 共 18576字 ⁄ 字号 小 中 大 ⁄ 评论关闭

 

 

连接池的结构类图已经在上面详细给出，现在我们来分析一个源码:

  1.      public GenericPool(PoolHelper helper, int minSize, int maxSize,
  2.             long lifeTime, long sleepTime, long maxLifeTime) {
  3.            //连接工厂
  4.         this.threadFactory = null;
  5.         //生命周期
  6.             this.lifeTime = lifeTime;
  7.         //最大生命周期
  8.             this.maxLifeTime = maxLifeTime;
  9.         //最小容量
  10.             this.minSize = minSize;
  11.         //最大容量
  12.             this.maxSize = maxSize;
  13.         //容器中包含对象的生产工厂
  14.             this.poolHelper = helper;
  15.             this.sleepTime = sleepTime;
  16.             this.checkLevelObject = 0;
  17.             this.deadLockMaxWait = DEFAULT_DEADLOCKMAXWAIT;
  18.             this.deadLockRetryWait = DEFAULT_DEADLOCKRETRYWAIT;
  19.         }

连接池的一些基本的参数设定；

  1. public synchronized void start() {
  2.         //已经在使用的对象队列
  3.         locked = new Hashtable(); // create locked objects pool
  4.         //当前可用的对象的队列
  5.         unlocked = new Hashtable(); // create unlocked objects pool
  6.         hitList = new Vector();
  7.         count = 0; // 0 element to start
  8.         gc = false; // do not actions concerning garbage collector
  9.         //记录起始时间，开始计算对象生成的时间；
  10.         long now = System.currentTimeMillis(); // current time
  12.         // to obtain to the beginning minSize objects in the pool
  13.         //生成最小的对象个数
  14.         for (int i = 0; i < minSize; i++) { // count have to be equal to minSize
  15.             try {
  16.                 //抽象工厂模式，自定义工厂生产自定义的产品；把具体的工厂和产品解耦；
  17.                 GenerationObject genObject = poolHelper.create();
  18.                 //把数据和时间都都存储起来
  19.                 unlocked.put(genObject, new Long(now));
  20.                 // put it in the unlocked pool
  21.             } catch (Exception e) {
  22.                 log.error("Error Exception in GenericPool:start " + e);
  23.             }
  24.             //把当前连接池所有对象的个数增加；
  25.             ++count; // there is one more element in the pool
  26.         }
  28.         // keeper removes dead or useless objects
  29.         if (threadFactory != null) {
  30.             try {
  31.                 this.poolKeeper = new PoolKeeper(sleepTime, this);
  32.                 this.keeper = threadFactory.getThread(poolKeeper);
  33.             } catch (Exception e) {
  34.                 throw new IllegalStateException(e.getMessage());
  35.             }
  36.         } else {
  37.             // keep a handle to the poolkeeper so we can destroy it later
  38.             this.poolKeeper = new PoolKeeper(sleepTime, this);
  39.             this.keeper = new Thread(poolKeeper);
  41.         }
  42.         //启动一个连接池维护线程；
  43.         keeper.start();
  44.         // start the thread to verify element in the pool(unlocked)
  45.         log.debug("GenericPool:start pool started");
  46.     }

在上面就形成了连接池的一个基本的容器；

  1. public GenerationObject create() throws SQLException {
  2.         return create(getUser(), getPassword());
  3.     }
  5.     public GenerationObject create(String _user, String _password)
  6.         throws SQLException {
  7.         log.debug(
  8.             "StandardPoolDataSource:create create a connection for the pool");
  9.         GenerationObject genObject;
  10.         PooledConnection pooledCon = cpds.getPooledConnection(_user, _password);
  11.         // get the pooled connection
  12.                  //生产的连接绑定到了自身，方便回笼；
  13.         pooledCon.addConnectionEventListener(this);
  14.         // add it to the event listener
  15.         log.debug("StandardPoolDataSource:create create a object for the pool");
  16.         genObject =
  17.             new GenerationObject(
  18.                 pooledCon,
  19.                 pool.getGeneration(),
  20.                 _user,
  21.                 _password);
  22.         return genObject; // return a connection
  23.     }

    下面介绍一下对象的取出和回笼；

  1. public Object checkOut(String user, String password) throws Exception {
  2.         log.debug("GenericPool:checkOut an object");
  3.         //拿到当前的时间；
  4.         long now = System.currentTimeMillis(); // current time to compare
  5.         GenerationObject o;
  6.         Enumeration e;
  7.         Object realObject;
  8.         log.debug("GenericPool:checkOut UnlockedObjectCount="
  9.                 + getUnlockedObjectCount());
  10.         log.debug("GenericPool:checkOut LockedObjectCount="
  11.                 + getLockedObjectCount());
  12.         log
  13.                 .debug("GenericPool:checkOut count=" + count + " maxSize="
  14.                         + maxSize);
  15.         //如果当前存在可用对象
  16.         if (getUnlockedObjectCount() > 0) {
  17.             // if there are objects in the unlocked pool
  18.             if ((checkLevelObject == 3) || (checkLevelObject == 4)) { // need
  19.                 // to
  20.                 // verify
  21.                 // all
  22.                 // the
  23.                 // objects
  24.                //遍历这些对象
  25.                 e = unlocked.keys();
  26.                 while (e.hasMoreElements()) {
  27.                     o = (GenerationObject) e.nextElement();
  28.                     //存在可用的对象
  29.                     realObject = o.getObj(); // take the current object
  30.                     // first, verify if the object is not dead (lifetime)
  31.                     //如果当前元素过期的话
  32.                     Long life = (Long) unlocked.get(o);
  33.                     if (life == null
  34.                         || (now - life.longValue()) > lifeTime
  35.                         || (maxLifeTime > 0 && (now - o.getCreated()) > maxLifeTime)) {
  36.                         
  37.                         // object has expired
  38.                         log.debug("GenericPool:checkOut an object has expired");
  39.                         //从可用的队列中移除
  40.                         removeUnlockedObject(o);
  41.                         // minimumObject(user, password);
  42.                         // build object in the pool if it is lesser than minSize
  43.                     } else {
  44.                         log
  45.                                 .debug("GenericPool:checkOut check the owner of the connection");
  46.                         //检查拥有者
  47.                         if (checkOwner(o, user, password)) {
  48.                             if (((checkLevelObject == 3) && !poolHelper
  49.                                     .checkThisObject(realObject))
  50.                                     || ((checkLevelObject == 4) && !poolHelper
  51.                                             .testThisObject(realObject))) {
  52.                                 log
  53.                                         .debug("GenericPool:checkOut remove object checkLevelObject="
  54.                                                 + checkLevelObject);
  55.                                 //不合法移除；
  56.                                 removeUnlockedObject(o);
  57.                                 // minimumObject(user, password);
  58.                                 // build object in the pool if it is lesser than
  59.                                 // minSize
  60.                             }
  61.                         }
  62.                     }
  63.                 }
  64.             }
  66.         }
  67. int currentWait = 0;
        
            Object obj = getFromPool(user, password);
            while ((obj == null) && (currentWait < getDeadLockMaxWait())) {
                log.info("GenericPool:checkOut waiting for an object :"
                        + this.poolHelper.toString());
                try {
                    synchronized (this) {
                        //当没有对象的时候，等待若干时间
                        wait(getDeadLockRetryWait());
                    }
                } catch (InterruptedException excp) {
                    log
                            .error("GenericPool:checkOut ERROR Failed while waiting for an object: "
                                    + excp);
                }
                currentWait += getDeadLockRetryWait();
                //在重新尝试从池中取出可用对象；
                obj = getFromPool(user, password);
            }

            if (obj == null)
                throw new Exception(
                        "GenericPool:checkOut ERROR  impossible to obtain a new object from the pool");

            return obj;
        }

  1. private Object getFromPool(String user, String password) throws Exception {
  3.         long now = System.currentTimeMillis(); // current time to compare
  4.         if (getUnlockedObjectCount() > 0) {
  5.             // now, we have to return an object to the user
  6.             GenerationObject o = null;
  7.             Object realObject = null;
  8.             Long life = null;
  10.             Enumeration e = unlocked.keys(); // then take them
  11.             while (e.hasMoreElements()) { // for each objects ...
  12.                 synchronized (this) {
  14.                     // Ensure that there are object in the unlocked
  15.                     // collection
  16.                     if (getUnlockedObjectCount() == 0)
  17.                         break;
  19.                     o = (GenerationObject) e.nextElement();
  20.                     life = (Long) unlocked.get(o);
  21.                     if (life == null) {
  22.                         // Fix for #303462; note that this fixes the problem, but Enumeration's on Hashtable's
  23.                         // are by definition somewhat unpredictable; a more robust fix may be in order
  24.                         log.debug("GenericPool:getFromPool fix for #303462 encountered");
  25.                         continue;
  26.                     }
  27.                     //从可用对象的队列中移除
  28.                     unlocked.remove(o);
  29.                     // In any case the object will be removed.
  30.                     // Prevents others accessing the object while we are
  31.                     // not synchronized.
  32.                     //返回其真实对象；
  33.                     realObject = o.getObj();
  34.                 }
  36.                 // Verify that the life object is valid
  37.                 if (life == null)
  38.                     break;
  40.                 // first, verify if the object is not dead (lifetime)
  41.                 if (life == null
  42.                     || (now - life.longValue()) > lifeTime
  43.                     || (maxLifeTime > 0 && (now - o.getCreated()) > maxLifeTime)) {
  44.                     
  45.                     // object has expired
  46.                     log.debug("GenericPool:getFromPool an object has expired");
  47.                     removeUnlockedObject(o);
  48.                 } else {
  49.                     log
  50.                             .debug("GenericPool:getFromPool check the owner of the connection");
  51.                     if (checkOwner(o, user, password)) {
  52.                         log.debug("GenericPool:getFromPool owner is verified");
  53.                         // second, verification of the object if needed
  54.                         if ((checkLevelObject == 0)
  55.                                 || ((checkLevelObject == 1) && poolHelper
  56.                                         .checkThisObject(realObject))
  57.                                 || ((checkLevelObject == 2) && poolHelper
  58.                                         .testThisObject(realObject))) {
  60.                             locked.put(o, new Long(now));
  61.                             // put it in the locked pool
  62.                             log
  63.                                     .debug("GenericPool:getFromPool return an object (after verification if needed)");
  64.                             return (o.getObj()); // return this element
  65.                         } else { // object failed validation
  66.                             //          System.out.println("removeUnlockedObject="+realObject);
  67.                             log
  68.                                     .debug("GenericPool:getFromPool kill an object from the pool");
  69.                             removeUnlockedObject(o);
  70.                         }
  72.                     } else
  73.                         log.debug("GenericPool:getFromPool owner is FALSE");
  74.                 }
  75.             }
  76.         } // if getUnlockedObjectCount() > 0
  78.         // if no objects available, create a new one
  79.         boolean create = false;
  80.         //若是已有对象总和没有超过最大容量，可以自己新建对象
  81.         synchronized (this) {
  82.             if (count < maxSize) {
  83.                 create = true;
  84.                 count++; // assume we can create a connection.
  85.             }
  86.         }
  87.         if (create) {
  88.             //          System.out.println("on doit creer une connection");
  89.             // if number of pooled object is < max size of the pool
  90.             log
  91.                     .debug("GenericPool:getFromPool no objects available, create a new one");
  92.             try {
  93.                 //          System.out.println("on doit creer une connection CREATE");
  94.                 GenerationObject genObject = poolHelper.create(user, password);
  95.                 //          System.out.println("nouvel objet="+genObject.getObj());
  96.                 //          System.out.println("on doit creer une connection PUT");
  97.                 locked.put(genObject, new Long(now));
  98.                 // put it in the locked pool
  99.                 return (genObject.getObj()); // and return this element
  100.             } catch (Exception excp) {
  101.                 synchronized (this) {
  102.                     count--; // our assumption failed. rollback.
  103.                 }
  105.                 log
  106.                         .error("GenericPool:getFromPool Error Exception in GenericPool:getFromPool");
  107.                 // cney: rethrow exception thrown by create
  108.                 throw excp;
  109.             }
  110.         }
  111.         //否则返回空，继续等待
  112.         return null;
  113.     }

从连接池中回笼；

  1. public synchronized void checkIn(Object o) {
  2.         log.debug("GenericPool:checkIn return an object to the pool");
  4.         for (Enumeration enumeration = locked.keys(); enumeration.hasMoreElements();) { // for
  5.             // each
  6.             // object
  7.             // of
  8.             GenerationObject obj = (GenerationObject) enumeration.nextElement();
  9.             // the locked pool
  11.             if (obj.getObj().equals(o)) {
  12.                 //从锁定的对象从锁定队列中移入可用队列中
  13.                 locked.remove(obj); // remove the object from the locked pool
  14.                 unlocked.put(obj, new Long(System.currentTimeMillis()));
  16.                 // we have to verify if the generation of the object is still
  17.                 // valid
  18.                 int genObj = obj.getGeneration(); // get the generation number
  20.                 // if the generation number of the object is not valid, test the
  21.                 // object
  22.                 if (generation > genObj) {
  23.                     if (!poolHelper.checkThisObject(obj.getObj()))
  24.                         // if the object is not valid
  25.                         removeUnlockedObject(obj);
  27.                 }
  28.                 //唤醒线程
  29.                 notifyAll();
  30.             }
  31.         }

我们再来看一下连接池的清理线程；

  1. public class PoolKeeper implements Runnable {
  3.     private long sleepTime; // timeout between 2 clean up
  4.     private GenericPool pool; // the pool to clean up
  5.     private boolean running = true;
  7.     /**
  8.      * constructor, called by GenericPool (any kind of object)
  9.      */
  10.     public PoolKeeper(long sleepTime, GenericPool pool) {
  11.         this.sleepTime = sleepTime;
  12.         this.pool = pool;
  13.     }
  15.     public void stop() {
  16.         synchronized (this ) {
  17.                         running = false;
  18.                 }
  19.     }
  21.     /**
  22.      * run method. allows to clean up the pool
  23.      */
  24.     public void run() {
  25.         while (running && !Thread.interrupted()) {
  26.             try {
  27.                 synchronized (this) {
  28.                     wait(this.sleepTime); // wait for timeout ms before attack
  29.                 }
  30.             } catch (InterruptedException e) {
  31.                                 break;
  32.             }
  33.                         //特定时间间隔的清扫连接池；
  35.             this.pool.cleanUp(); // clean up the Pool and reallocate objects
  36.         }
  37.         // release the pool.
  38.         this.pool = null;
  39.     }

在clearUp中：

  1. long now = System.currentTimeMillis(); // current time
  2.         synchronized (this) {
  3.             for (Enumeration enumeration = unlocked.keys(); enumeration.hasMoreElements();) { // for
  4.                 // each
  5.                 // object
  6.                 // of
  7.                 // the
  8.                 GenerationObject o = (GenerationObject) enumeration.nextElement();
  9.                 // unlocked pool
  10.                 
  11.                 Long lasttouch = (Long) unlocked.get(o);
  12.                 // birth day of the pool
  13.                 if (lasttouch == null
  14.                     || (now - lasttouch.longValue()) > lifeTime
  15.                     || (maxLifeTime > 0 && (now - o.getCreated()) > maxLifeTime)) {
  16.                     log.debug("GenericPool:cleanUp clean up the pool");
  17.                     removeUnlockedObject(o);
  18.                 }
  19.             }
  20.         }

用当前时间与创建时间的间隔和生命周期做比较，看是否过期，过期杀死；

  1. public void expire(Object o) {
  2.         log.debug(
  3.             "StandardPoolDataSource:expire expire a connection, remove from the pool");
  4.         if (o == null)
  5.             return;
  6.         try {
  7.             PooledConnection pooledCon = (PooledConnection) o;
  8.             pooledCon.close(); // call close() of PooledConnection
  9.             pooledCon.removeConnectionEventListener(this);
  10.             log.debug("StandardPoolDataSource:expire close the connection");
  11.         } catch (java.sql.SQLException e) {
  12.             log.error(
  13.                 "StandardPoolDataSource:expire Error java.sql.SQLException in StandardPoolDataSource:expire");
  14.         }
  15.     }

我们再来看一下数据源的如何获取连接；

  1. public Connection getConnection(String _user, String _password)
  2.         throws SQLException {
  3.         log.debug("StandardPoolDataSource:getConnection");
  4.         Connection ret = null;
  5.         PooledConnection con = null;
  7.         synchronized (this) {
  8.             if (!onOff) {
  9.                 log.debug(
  10.                     "StandardPoolDataSource:getConnection must configure the pool...");
  11.                 pool.start(); // the pool starts now
  12.                 onOff = true; // and is initialized
  13.                 log.debug(
  14.                     "StandardPoolDataSource:getConnection pool config : /n"
  15.                         + pool.toString());
  16.             }
  17.         }
  19.         try {
  20.             try {
  21.                 log.debug(
  22.                     "StandardPoolDataSource:getConnection Try to give a "
  23.                         + "connection (checkOut)");
  24.                 //从连接池中取得PooledConnection；
  25.                 con = (PooledConnection) pool.checkOut(_user, _password);
  26.                 // get a connection from the pool
  27.                 log.debug(
  28.                     "StandardPoolDataSource:getConnection checkOut return"
  29.                         + "a new connection");
  30.             } catch (Exception e) {
  31.                                 e.printStackTrace();
  32.                 log.debug(
  33.                     "StandardPoolDataSource:getConnection SQLException in StandardPoolDataSource:getConnection"
  34.                         + e);
  35.                 throw new SQLException(
  36.                     "SQLException in StandardPoolDataSource:getConnection no connection available "
  37.                         + e);
  38.             }
  39.             //取得物理连接
  40.             ret = con.getConnection();
  41.         } catch (Exception e) {
  42.             log.debug("StandardPoolDataSource:getConnection exception" + e);
  43.                         e.printStackTrace();
  44.             SQLException sqle =
  45.                 new SQLException(
  46.                     "SQLException in StandardPoolDataSource:getConnection exception: "
  47.                         + e);
  48.             if (e instanceof SQLException)
  49.                 sqle.setNextException((SQLException) e);
  50.             if (con != null) {
  51.                 pool.checkIn(con);
  52.             }
  53.             throw sqle;
  54.         }
  55.         log.debug("StandardPoolDataSource:getConnection return a connection");
  56.         return ret;
  57.     }

分布式事务的分析会留到下一节，在这一节中，有几个地方值得学习：

（1）工厂和监听者集为一身，在场景，如产品还要工厂进行同意的售后服务等等；

 (2)线程扫描，时间间隔等等；

返回
【上篇】
【下篇】

作者:

抱歉!评论已关闭.

返回首页

Copyright © 2013-2018 学步园  保留所有权利.
软文销售 QQ客服：2265327166 点击这里给我发消息（其他合作也可洽谈）
必威体育
必威电竞