GenericObjectPool

commons-pool2源碼走讀(四) 對象池實現GenericObjectPool

2018年05月27日 18:24:56 藍墨49 閱讀數 1787

 

版權聲明：本文為博主原創文章，遵循 CC 4.0 by-sa 版權協議，轉載請附上原文出處鏈接和本聲明。

本文鏈接： https://blog.csdn.net/qq447995687/article/details/80469790

commons-pool2源碼走讀(四) 對象池實現GenericObjectPool<T>

GenericObjectPool <T> 是一個可配置的ObjectPool實現。 
當與適當的PooledObjectFactory組合使用時，GenericObjectPool為任意對象提供健壯的池功能。

您可以選擇性的配置池來檢查和可能回收池中的空閑對象，並確保有最少數量的空閑對象可用。這是由一個“空閑對象回收”線程（即BaseGenericObjectPool <T> 的Evictor）執行的，線程是異步運行的。在配置這個可選特性時，應該謹慎使用。驅逐運行與客戶端線程爭用池中的對象，因此如果它們運行得太頻繁，可能會導致性能問題。

還可以配置池來檢測和刪除被泄漏的對象，比如一個從池中借出的對象，在超過removeAbandonedTimeout超時之前既不使用也不返回。移除泄漏的連接，可能發生在對象被借用時對象池已接近飽和，也可能是被回收線程檢查出，或者兩者都執行時。如果池對象實現了TrackedUse接口，那么其最后一次使用時間使取決於getLastUsed方法;否則，是由對象從池中借出的時間決定。

實現注意:為了防止可能的死鎖，已經采取了謹慎措施，以確保在同步塊中不會發生對工廠方法的調用。這個類線程安全。

1、接口繼承、實現關系

GenericObjectPool <T> 實現了ObjectPool<T> 具備對象池的功能，同時 繼承了BaseGenericObjectPool<T> 的對於對象狀態管理和回收等功能。 

2、構造函數

構造函數通過GenericObjectPoolConfig 和PooledObjectFactory來進行參數的初始化和對象工廠類的引入。

    public GenericObjectPool(final PooledObjectFactory<T> factory, final GenericObjectPoolConfig config) { //父類BaseGenericObjectPool構造方法 super(config, ONAME_BASE, config.getJmxNamePrefix()); if (factory == null) { jmxUnregister(); // tidy up throw new IllegalArgumentException("factory may not be null"); } this.factory = factory; //空閑對象隊列，此隊列非JDK而是自行實現的一個隊列 idleObjects = new LinkedBlockingDeque<>(config.getFairness()); //覆蓋BaseGenericObjectPool里面的配置參數 setConfig(config); //初始化回收線程 startEvictor(getTimeBetweenEvictionRunsMillis()); }

3、相關屬性

    // --- 可配置的屬性 ------------------------------------------------- //最大空閑數量 private volatile int maxIdle = GenericObjectPoolConfig.DEFAULT_MAX_IDLE; //最小空閑數量 private volatile int minIdle = GenericObjectPoolConfig.DEFAULT_MIN_IDLE; //對象工廠 private final PooledObjectFactory<T> factory; // --- 內部屬性 ------------------------------------------------- //池中所有的對象，只能是<=maxActive private final Map<IdentityWrapper<T>, PooledObject<T>> allObjects = new ConcurrentHashMap<>(); //已創建對象總數(不包含已銷毀的) private final AtomicLong createCount = new AtomicLong(0); //調用創建方法總線程數 private long makeObjectCount = 0; //makeObjectCount 增長時並發鎖 private final Object makeObjectCountLock = new Object(); //空閑對象隊列 private final LinkedBlockingDeque<PooledObject<T>> idleObjects; // JMX specific attributes private static final String ONAME_BASE = "org.apache.commons.pool2:type=GenericObjectPool,name="; //泄漏對象回收配置參數 private volatile AbandonedConfig abandonedConfig = null;

4、 對象池方法實現

• 借用對象 
  整個流程為，檢查池是否關閉 –> 是否回收泄漏對象 –> 是否阻塞創建對象 –> 創建對象 –> 分配對象 –> 激活對象 –> 校驗對象 –> 更改借用信息 –> 返回對象

   public T borrowObject(final long borrowMaxWaitMillis) throws Exception { //判斷對象池是否關閉：BaseGenericObjectPool.closed==true assertOpen(); //如果回收泄漏的參數配置不為空，並且removeAbandonedOnBorrow參數配置為true //並且Idle數量<2,Active數量>總數Total-3 //在借用時進行回收泄漏連接（會影響性能） final AbandonedConfig ac = this.abandonedConfig; if (ac != null && ac.getRemoveAbandonedOnBorrow() && (getNumIdle() < 2) && (getNumActive() > getMaxTotal() - 3) ) { //回收泄漏對象 removeAbandoned(ac); } PooledObject<T> p = null; //copy blockWhenExhausted 防止其它線程更改getBlockWhenExhausted值造成並發問題 //借用對象時如果沒有是否阻塞直到有對象產生 final boolean blockWhenExhausted = getBlockWhenExhausted(); //創建成功標識 boolean create; //記錄當前時間，用作記錄借用操作總共花費的時間 final long waitTime = System.currentTimeMillis(); //當對象為空時一直獲取 while (p == null) { create = false; //從雙端隊列彈出第一個隊首對象，為空返回null p = idleObjects.pollFirst(); //如果為空則重新創建一個對象 if (p == null) { //創建對象 p = create(); //p==null可能對象池達到上限不能繼續創建！ if (p != null) { create = true; } } //如果對象p還是為空則阻塞等待 if (blockWhenExhausted) { if (p == null) { if (borrowMaxWaitMillis < 0) { //沒有超時時間則阻塞等待到有對象為止 p = idleObjects.takeFirst(); } else { //有超時時間 p = idleObjects.pollFirst(borrowMaxWaitMillis, TimeUnit.MILLISECONDS); } } //達到超時時間，還未取到對象，則拋出異常 if (p == null) { throw new NoSuchElementException( "Timeout waiting for idle object"); } } else { //未取到對象，則拋出異常 if (p == null) { throw new NoSuchElementException("Pool exhausted"); } } //調用PooledObject.allocate()方法分配對象 //[具體實現請看](https://blog.csdn.net/qq447995687/article/details/80413227) if (!p.allocate()) { p = null; } //分配成功 if (p != null) { try { //激活對象，具體請看factory實現，對象重借出到歸還整個流程經歷的過程圖 factory.activateObject(p); } catch (final Exception e) { try { destroy(p); } catch (final Exception e1) { // Ignore - activation failure is more important } p = null; if (create) { final NoSuchElementException nsee = new NoSuchElementException( "Unable to activate object"); nsee.initCause(e); throw nsee; } } //對象創建成功，是否進行測試 if (p != null && (getTestOnBorrow() || create && getTestOnCreate())) { boolean validate = false; Throwable validationThrowable = null; try { //校驗對象，具體請看factory實現，對象重借出到歸還整個流程經歷的過程圖 validate = factory.validateObject(p); } catch (final Throwable t) { PoolUtils.checkRethrow(t); validationThrowable = t; } //校驗不通過則銷毀對象 if (!validate) { try { destroy(p); destroyedByBorrowValidationCount.incrementAndGet(); } catch (final Exception e) { // Ignore - validation failure is more important } p = null; if (create) { final NoSuchElementException nsee = new NoSuchElementException( "Unable to validate object"); nsee.initCause(validationThrowable); throw nsee; } } } } } //更新對象借用狀態 updateStatsBorrow(p, System.currentTimeMillis() - waitTime); return p.getObject(); }

創建對象 
當借用時，空閑對象為空，並且未達到池最大數量，則會調用該方法重新創建一個空閑對象

   private PooledObject<T> create() throws Exception { int localMaxTotal = getMaxTotal(); // 如果最大數量為負數則設置為Integer的最大值 if (localMaxTotal < 0) { localMaxTotal = Integer.MAX_VALUE; } // 是否創建成功的一個flag: // - TRUE: 調用工廠類成功創建一個對象 // - FALSE: 返回空 // - null: 重復創建 Boolean create = null; while (create == null) { synchronized (makeObjectCountLock) { //加上本次操作，總共創建個數 final long newCreateCount = createCount.incrementAndGet(); if (newCreateCount > localMaxTotal) { //連接池容量已滿，不能繼續增長。在對最后一個對象的創建上， //加入了設計較為精妙，需細細揣摩 createCount.decrementAndGet(); //調用創建對象方法線程數=0 if (makeObjectCount == 0) { //容量已滿並且沒有線程調用makeObject()方法， //表明沒有任何可能性再繼續創建對象， //返回並等待歸還的空閑對象 create = Boolean.FALSE; } else { //其它線程調用makeObject()方法在創建對象了。 //如果繼續創建則可能超過對象池容量，不返回false，因為其它線程也在創建， //但是是否能夠創建成功是未知的，如果其它線程沒能創建成功， //則此線程可能會搶奪到繼續創建的權利。 //釋放鎖，等待其它線程創建結束並喚醒該線程 makeObjectCountLock.wait(); } } else { // 對象池未滿，從新創建一個對象 makeObjectCount++; create = Boolean.TRUE; } } } //對象池容量達到上限，返回null重新等待其它線程歸還對象 if (!create.booleanValue()) { return null; } final PooledObject<T> p; try { //創建一個新對象 p = factory.makeObject(); } catch (final Exception e) { createCount.decrementAndGet(); throw e; } finally { //與上面wait()方法相呼應， //如果上面拋出了異常，喚醒其它線程爭奪繼續創建最后一個資源的權利 synchronized (makeObjectCountLock) { makeObjectCount--; makeObjectCountLock.notifyAll(); } } //設置泄漏參數，並加入調用堆棧 final AbandonedConfig ac = this.abandonedConfig; if (ac != null && ac.getLogAbandoned()) { p.setLogAbandoned(true); // TODO: in 3.0, this can use the method defined on PooledObject if (p instanceof DefaultPooledObject<?>) { ((DefaultPooledObject<T>) p).setRequireFullStackTrace(ac.getRequireFullStackTrace()); } } //將創建總數增加，並將對象放入 allObjects createdCount.incrementAndGet(); allObjects.put(new IdentityWrapper<>(p.getObject()), p); return p; }

回收泄漏對象

 private void removeAbandoned(final AbandonedConfig ac) { // Generate a list of abandoned objects to remove final long now = System.currentTimeMillis(); //超時時間=當前時間-配置的超時時間，如果一個對象的上次借用時間在此時間之前， //說明上次借用后經過了removeAbandonedTimeout時間限制還未被歸還過即可能是泄漏的對象 final long timeout = now - (ac.getRemoveAbandonedTimeout() * 1000L); //泄漏的，需要移除的對象列表 final ArrayList<PooledObject<T>> remove = new ArrayList<>(); final Iterator<PooledObject<T>> it = allObjects.values().iterator(); //遍歷池中對象依次判斷是否需要移除 while (it.hasNext()) { final PooledObject<T> pooledObject = it.next(); synchronized (pooledObject) { //如果對象的狀態為已分配ALLOCATED ，並且已經超過泄漏定義時間則添加到需要移除隊列進行統一移除 if (pooledObject.getState() == PooledObjectState.ALLOCATED && pooledObject.getLastUsedTime() <= timeout) { pooledObject.markAbandoned(); remove.add(pooledObject); } } } // 移除泄漏連接，如果配置了打印堆棧，則打印調用堆棧信息 final Iterator<PooledObject<T>> itr = remove.iterator(); while (itr.hasNext()) { final PooledObject<T> pooledObject = itr.next(); if (ac.getLogAbandoned()) { pooledObject.printStackTrace(ac.getLogWriter()); } try { //銷毀對象 invalidateObject(pooledObject.getObject()); } catch (final Exception e) { e.printStackTrace(); } } }

• invalidate對象

    public void invalidateObject(final T obj) throws Exception { //從所有對象中取出該對象，如果不存在則拋出異常 final PooledObject<T> p = allObjects.get(new IdentityWrapper<>(obj)); if (p == null) { if (isAbandonedConfig()) { return; } throw new IllegalStateException( "Invalidated object not currently part of this pool"); } //如果對象不是無效狀態PooledObjectState.INVALID，則銷毀此對象 synchronized (p) { if (p.getState() != PooledObjectState.INVALID) { destroy(p); } } // ensureIdle(1, false); }

銷毀對象 
銷毀對象，並從池中移除，更新對象池已創建數量和總銷毀數量

    private void destroy(final PooledObject<T> toDestroy) throws Exception { toDestroy.invalidate(); idleObjects.remove(toDestroy); allObjects.remove(new IdentityWrapper<>(toDestroy.getObject())); try { factory.destroyObject(toDestroy); } finally { destroyedCount.incrementAndGet(); createCount.decrementAndGet(); } }

確保最小空閑數量

    private void ensureIdle(final int idleCount, final boolean always) throws Exception { //!idleObjects.hasTakeWaiters()如果idleObjects隊列還有線程等待獲取對象則由最后一個 //等待者確保最小空閑數量 if (idleCount < 1 || isClosed() || (!always && !idleObjects.hasTakeWaiters())) { return; } //一直創建空閑對象知道空閑對象數量>總空閑數量閾值 while (idleObjects.size() < idleCount) { final PooledObject<T> p = create(); if (p == null) { // Can't create objects, no reason to think another call to // create will work. Give up. break; } //根據先進先出參數，添加對象到隊首或者隊尾 if (getLifo()) { idleObjects.addFirst(p); } else { idleObjects.addLast(p); } } //在此過程中如果連接池關閉則clear所有對象 if (isClosed()) { clear(); } }

• 歸還對象

歸還對象方法將適用完的對象從新放置回對象池中重復利用。其整個流程為：檢查是否存在 –> 檢查狀態是否正確 –> 是否在歸還時測試對象 –> 校驗對象 –> 鈍化（卸載）對象 –> 結束分配 –> 銷毀/歸還該對象 –> 更新連接池歸還信息

    public void returnObject(final T obj) { final PooledObject<T> p = allObjects.get(new IdentityWrapper<>(obj)); if (p == null) { //如果對象為空，並且沒有配置泄漏參數則拋出異常，表明該對象不是連接池中的對象 if (!isAbandonedConfig()) { throw new IllegalStateException( "Returned object not currently part of this pool"); } //如果對象為空，表明該對象是abandoned並且已被銷毀 return; } synchronized(p) { final PooledObjectState state = p.getState(); //如果被歸還的對象不是已分配狀態，拋出異常 if (state != PooledObjectState.ALLOCATED) { throw new IllegalStateException( "Object has already been returned to this pool or is invalid"); } //更改狀態為returning，避免在此過程中被標記為被遺棄。 p.markReturning(); } final long activeTime = p.getActiveTimeMillis(); //是否在歸還時測試該對象 if (getTestOnReturn()) { //校驗對象 if (!factory.validateObject(p)) { try { //校驗不通過則destroy對象 destroy(p); } catch (final Exception e) { swallowException(e); } try { //確保最小空閑數量 ensureIdle(1, false); } catch (final Exception e) { swallowException(e); } //更新連接池歸還信息BaseGenericObjectPool#returnedCount,activeTimes updateStatsReturn(activeTime); return; } } //校驗通過 try { //鈍化（卸載）對象 factory.passivateObject(p); } catch (final Exception e1) { swallowException(e1); try { destroy(p); } catch (final Exception e) { swallowException(e); } try { ensureIdle(1, false); } catch (final Exception e) { swallowException(e); } updateStatsReturn(activeTime); return; } //結束分配，如果對象為ALLOCATED或者RETURNING更改對象為空閑IDLE狀態 //具體看org.apache.commons.pool2.impl.DefaultPooledObject#deallocate方法 if (!p.deallocate()) { throw new IllegalStateException( "Object has already been returned to this pool or is invalid"); } final int maxIdleSave = getMaxIdle(); //如果對象池已經關閉或者空閑數量達到上限，則銷毀該對象 if (isClosed() || maxIdleSave > -1 && maxIdleSave <= idleObjects.size()) { try { destroy(p); } catch (final Exception e) { swallowException(e); } } else { //否則將歸還的對象添加到空閑隊列，連接池的最終目的：重用一個連接 if (getLifo()) { idleObjects.addFirst(p); } else { idleObjects.addLast(p); } if (isClosed()) { // Pool closed while object was being added to idle objects. // Make sure the returned object is destroyed rather than left // in the idle object pool (which would effectively be a leak) clear(); } } updateStatsReturn(activeTime); }

• clear連接池 
  依次銷毀每個鏈接

    public void clear() { PooledObject<T> p = idleObjects.poll(); while (p != null) { try { destroy(p); } catch (final Exception e) { swallowException(e); } p = idleObjects.poll(); } }

• 回收對象

此方法實現了org.apache.commons.pool2.impl.BaseGenericObjectPool#evict 方法，用於回收線程回收空閑對象。 
回收的整個流程為：判斷池是否關閉及是否有空閑對象 –> 根據策略獲得回收的條數 –> 判斷對象狀態開始進行回收 –> 根據回收策略EvictionPolicy判斷是否能夠回收 –> 如能回收則銷毀對象 –> 不能回收則判斷是否校驗對象 –> 激活對象 –> 校驗對象 –> 鈍化對象 –> 結束回收更改對象狀態 –> 回收泄漏連接

public void evict() throws Exception { assertOpen(); if (idleObjects.size() > 0) { PooledObject<T> underTest = null; final EvictionPolicy<T> evictionPolicy = getEvictionPolicy(); synchronized (evictionLock) { //回收參數 final EvictionConfig evictionConfig = new EvictionConfig( getMinEvictableIdleTimeMillis(), getSoftMinEvictableIdleTimeMillis(), getMinIdle()); //是否在回收時測試對象 final boolean testWhileIdle = getTestWhileIdle(); //根據getNumTests()對部分對象進行回收測試 for (int i = 0, m = getNumTests(); i < m; i++) { //evictionIterator是空閑對象的一個迭代器,可以想象為idleObjects.iterator() if (evictionIterator == null || !evictionIterator.hasNext()) { evictionIterator = new EvictionIterator(idleObjects); } if (!evictionIterator.hasNext()) { // Pool exhausted, nothing to do here return; } //多線程並發時，有可能上面檢測到有對象，而另一個對象隨后將其借出 try { underTest = evictionIterator.next(); } catch (final NoSuchElementException nsee) { // 對象被其它線程借出 i--; evictionIterator = null; continue; } //根據狀態判斷是否能夠開始回收測試，並更改狀態，詳細實現請看源碼走讀(一) if (!underTest.startEvictionTest()) { // Object was borrowed in another thread // Don't count this as an eviction test so reduce i; i--; continue; } //根據回收策略判斷對象是否能夠被回收，單獨分析 boolean evict; try { //根據回收策略判斷對象是否能夠被回收 evict = evictionPolicy.evict(evictionConfig, underTest, idleObjects.size()); } catch (final Throwable t) { // Slightly convoluted as SwallowedExceptionListener // uses Exception rather than Throwable PoolUtils.checkRethrow(t); swallowException(new Exception(t)); // Don't evict on error conditions evict = false; } //如果能被回收則銷毀對象 if (evict) { destroy(underTest); destroyedByEvictorCount.incrementAndGet(); } else { //不能被回收，則是否進行校驗，與借出流程相同，只不過該處只是校驗而沒有借出實際使用 if (testWhileIdle) { boolean active = false; try { //對象已經被借出，直接激活 factory.activateObject(underTest); active = true; } catch (final Exception e) { destroy(underTest); destroyedByEvictorCount.incrementAndGet(); } if (active) { //激活成功進行校驗 if (!factory.validateObject(underTest)) { //校驗不通過則銷毀對象 destroy(underTest); destroyedByEvictorCount.incrementAndGet(); } else { try { //校驗通過則重新將對象鈍化(卸載) factory.passivateObject(underTest); } catch (final Exception e) { destroy(underTest); destroyedByEvictorCount.incrementAndGet(); } } } } //結束回收測試，更改對象狀態或者添加到空閑隊列， //如果在此途中被借出，還需重新添加到idleObjects，具體實現請看源碼走讀(一) if (!underTest.endEvictionTest(idleObjects)) { // TODO - May need to add code here once additional // states are used } } } } } //配置了回收，則進行回收泄漏連接 final AbandonedConfig ac = this.abandonedConfig; if (ac != null && ac.getRemoveAbandonedOnMaintenance()) { removeAbandoned(ac); } }

返回有多少對象需要進行回收測試

    private int getNumTests() { final int numTestsPerEvictionRun = getNumTestsPerEvictionRun(); if (numTestsPerEvictionRun >= 0) { return Math.min(numTestsPerEvictionRun, idleObjects.size()); } return (int) (Math.ceil(idleObjects.size() / Math.abs((double) numTestsPerEvictionRun))); }