使用Java實作一個通用並發物件池的程式碼分享
- 黄舟原創
- 2017-03-24 10:55:551250瀏覽
這篇文章裡我們主要討論下如何在Java裡實作一個物件池。最近幾年，Java虛擬機的效能在各方面都得到了極大的提升，因此對大多數物件而言，已經沒有必要透過物件池來提高效能了。根本的原因是，創建一個新的物件的開銷已經不像過去那麼昂貴了。
然而，還是有些對象，它們的創建開銷是非常大的，比如線程，資料庫連接等這些非輕量級的對象。在任何一個應用程式裡面，我們一定會用到不只一個這樣的物件。如果有一個很方便的創建管理這些物件的池，使得這些物件能夠動態的重用，而客戶端程式碼也不關心它們的生命週期，還是會很給力的。
在真正開始寫程式碼前，我們先來梳理下一個物件池需要完成哪些功能。
如果有可用的對象，則物件池應能傳回給客戶端。
客戶端把物件放回池裡後，可以對這些物件進行重複使用。
物件池能夠建立新的物件來滿足客戶端不斷增長的需求。
需要有一個正確關閉池的機制來確保關閉後不會發生記憶體外洩。
不用說了，上面幾點就是我們要暴露給客戶端的連接池的介面的基本功能。
我們的宣告的介面如下：
package com.test.pool;
/**
* Represents a cached pool of objects.
*
* @author Swaranga
*
* @param <T> the type of object to pool.
*/
public interface Pool<T>
{
/**
* Returns an instance from the pool.
* The call may be a blocking one or a non-blocking one
* and that is determined by the internal implementation.
*
* If the call is a blocking call,
* the call returns immediately with a valid object
* if available, else the thread is made to wait
* until an object becomes available.
* In case of a blocking call,
* it is advised that clients react
* to {@link InterruptedException} which might be thrown
* when the thread waits for an object to become available.
*
* If the call is a non-blocking one,
* the call returns immediately irrespective of
* whether an object is available or not.
* If any object is available the call returns it
* else the call returns < code >null< /code >.
*
* The validity of the objects are determined using the
* {@link Validator} interface, such that
* an object < code >o< /code > is valid if
* < code > Validator.isValid(o) == true < /code >.
*
* @return T one of the pooled objects.
*/
T get();
/**
* Releases the object and puts it back to the pool.
*
* The mechanism of putting the object back to the pool is
* generally asynchronous,
* however future implementations might differ.
*
* @param t the object to return to the pool
*/
void release(T t);
/**
* Shuts down the pool. In essence this call will not
* accept any more requests
* and will release all resources.
* Releasing resources are done
* via the < code >invalidate()< /code >
* method of the {@link Validator} interface.
*/
void shutdown();
}為了能夠支援任意對象，上面這個介面故意設計得很簡單通用。它提供了從池中獲取/返回物件的方法，還有一個關閉池的機制，以便釋放物件。
現在我們來實作一下這個介面。在開始動手之前，值得一提的是，一個理想的release方法應該先試著檢查下這個客戶端回傳的物件是否還能重複使用。如果是的話再把它丟回池裡，如果不是，就捨棄掉這個對象。我們希望這個Pool介面的所有實作都能遵循這個規則。在開始具體的實作類別之前，我們先建立一個抽象類別，以便限制後續的實作能遵循這一點。我們實作的抽象類別就叫做AbstractPool，它的定義如下：
package com.test.pool;
/**
* Represents an abstract pool, that defines the procedure
* of returning an object to the pool.
*
* @author Swaranga
*
* @param <T> the type of pooled objects.
*/
abstract class AbstractPool <T> implements Pool <T>
{
/**
* Returns the object to the pool.
* The method first validates the object if it is
* re-usable and then puts returns it to the pool.
*
* If the object validation fails,
* some implementations
* will try to create a new one
* and put it into the pool; however
* this behaviour is subject to change
* from implementation to implementation
*
*/
@Override
public final void release(T t)
{
if(isValid(t))
{
returnToPool(t);
}
else
{
handleInvalidReturn(t);
}
}
protected abstract void handleInvalidReturn(T t);
protected abstract void returnToPool(T t);
protected abstract boolean isValid(T t);
}在上面這個類別裡，我們讓物件池必須得先驗證物件後才能把它放回池裡。具體的實作可以自由選擇如何實現這三種方法，以便自訂自己的行為。它們會根據自己的邏輯來決定如何判斷一個物件有效，無效的話該怎麼處理（handleInvalidReturn方法），怎麼把一個有效的物件放回池裡（returnToPool方法）。
有了上面這幾個類，我們就可以著手開始具體的實作了。不過還有個問題，由於上面這些類別是設計成能支援通用的物件池的，因此具體的實作不知道該如何驗證物件的有效性（因為物件都是泛型的）。因此我們還需要些別的東西來幫助我們完成這件事。
我們需要一個通用的方法來完成物件的校驗，而具體的實作不必關心物件是何種類型。因此我們引入了一個新的接口，Validator，它定義了驗證對象的方法。這個介面的定義如下：
package com.test.pool;
/**
* Represents the functionality to
* validate an object of the pool
* and to subsequently perform cleanup activities.
*
* @author Swaranga
*
* @param < T > the type of objects to validate and cleanup.
*/
public static interface Validator < T >
{
/**
* Checks whether the object is valid.
*
* @param t the object to check.
*
* @return <code>true</code>
* if the object is valid else <code>false</code>.
*/
public boolean isValid(T t);
/**
* Performs any cleanup activities
* before discarding the object.
* For example before discarding
* database connection objects,
* the pool will want to close the connections.
* This is done via the
* <code>invalidate()</code> method.
*
* @param t the object to cleanup
*/
public void invalidate(T t);
}上面這個介面定義了一個檢驗物件的方法，以及一個把物件置為無效的方法。當準備廢棄一個物件並清理記憶體的時候，invalidate方法就派上用場了。值得注意的是這個介面本身沒有任何意義，只有當它在物件池裡使用的時候才有意義，所以我們把這個介面定義到Pool介面裡面。這和Java集合庫裡的Map和Map.Entry是一樣的。所以我們的Pool介面就變成這樣了：
package com.test.pool;
/**
* Represents a cached pool of objects.
*
* @author Swaranga
*
* @param < T > the type of object to pool.
*/
public interface Pool< T >
{
/**
* Returns an instance from the pool.
* The call may be a blocking one or a non-blocking one
* and that is determined by the internal implementation.
*
* If the call is a blocking call,
* the call returns immediately with a valid object
* if available, else the thread is made to wait
* until an object becomes available.
* In case of a blocking call,
* it is advised that clients react
* to {@link InterruptedException} which might be thrown
* when the thread waits for an object to become available.
*
* If the call is a non-blocking one,
* the call returns immediately irrespective of
* whether an object is available or not.
* If any object is available the call returns it
* else the call returns < code >null< /code >.
*
* The validity of the objects are determined using the
* {@link Validator} interface, such that
* an object < code >o< /code > is valid if
* < code > Validator.isValid(o) == true < /code >.
*
* @return T one of the pooled objects.
*/
T get();
/**
* Releases the object and puts it back to the pool.
*
* The mechanism of putting the object back to the pool is
* generally asynchronous,
* however future implementations might differ.
*
* @param t the object to return to the pool
*/
void release(T t);
/**
* Shuts down the pool. In essence this call will not
* accept any more requests
* and will release all resources.
* Releasing resources are done
* via the < code >invalidate()< /code >
* method of the {@link Validator} interface.
*/
void shutdown();
/**
* Represents the functionality to
* validate an object of the pool
* and to subsequently perform cleanup activities.
*
* @author Swaranga
*
* @param < T > the type of objects to validate and cleanup.
*/
public static interface Validator < T >
{
/**
* Checks whether the object is valid.
*
* @param t the object to check.
*
* @return <code>true</code>
* if the object is valid else <code>false</code>.
*/
public boolean isValid(T t);
/**
* Performs any cleanup activities
* before discarding the object.
* For example before discarding
* database connection objects,
* the pool will want to close the connections.
* This is done via the
* <code>invalidate()</code> method.
*
* @param t the object to cleanup
*/
public void invalidate(T t);
}
}準備工作已經差不多了，在最後開始前我們還需要一個終極武器，這才是這個物件池的殺手鐧。就是「能夠創建新的物件」。我們的物件池是泛型的，因此它們得知道如何去產生新的物件來填充這個池子。這個功能不能依賴物件池本身，必須要有一個通用的方式來建立新的物件。透過一個ObjectFactory的介面就能完成這個，它只有一個「如何創建新的物件」的方法。我們的ObjectFactory介面如下：
package com.test.pool;
/**
* Represents the mechanism to create
* new objects to be used in an object pool.
*
* @author Swaranga
*
* @param < T > the type of object to create.
*/
public interface ObjectFactory < T >
{
/**
* Returns a new instance of an object of type T.
*
* @return T an new instance of the object of type T
*/
public abstract T createNew();
}我们的工具类都已经搞定了，现在可以开始真正实现我们的Pool接口了。因为我们希望这个池能在并发程序里面使用，所以我们会创建一个阻塞的对象池，当没有对象可用的时候，让客户端先阻塞住。我们的阻塞机制是让客户端一直阻塞直到有对象可用为止。这样的话导致我们还需要再增加一个只阻塞一定时间的方法，如果在超时时间到来前有对象可用则返回，如果超时了就返回null而不是一直等待下去。这样的实现有点类似Java并发库里的LinkedBlockingQueue，因此真正实现前我们再暴露一个接口，BlockingPool，类似于Java并发库里的BlockingQueue接口。
这里是BlockingQueue的声明：
package com.test.pool;
import java.util.concurrent.TimeUnit;
/**
* Represents a pool of objects that makes the
* requesting threads wait if no object is available.
*
* @author Swaranga
*
* @param < T > the type of objects to pool.
*/
public interface BlockingPool < T > extends Pool < T >
{
/**
* Returns an instance of type T from the pool.
*
* The call is a blocking call,
* and client threads are made to wait
* indefinitely until an object is available.
* The call implements a fairness algorithm
* that ensures that a FCFS service is implemented.
*
* Clients are advised to react to InterruptedException.
* If the thread is interrupted while waiting
* for an object to become available,
* the current implementations
* sets the interrupted state of the thread
* to <code>true</code> and returns null.
* However this is subject to change
* from implementation to implementation.
*
* @return T an instance of the Object
* of type T from the pool.
*/
T get();
/**
* Returns an instance of type T from the pool,
* waiting up to the
* specified wait time if necessary
* for an object to become available..
*
* The call is a blocking call,
* and client threads are made to wait
* for time until an object is available
* or until the timeout occurs.
* The call implements a fairness algorithm
* that ensures that a FCFS service is implemented.
*
* Clients are advised to react to InterruptedException.
* If the thread is interrupted while waiting
* for an object to become available,
* the current implementations
* set the interrupted state of the thread
* to <code>true</code> and returns null.
* However this is subject to change
* from implementation to implementation.
*
*
* @param time amount of time to wait before giving up,
* in units of <tt>unit</tt>
* @param unit a <tt>TimeUnit</tt> determining
* how to interpret the
* <tt>timeout</tt> parameter
*
* @return T an instance of the Object
* of type T from the pool.
*
* @throws InterruptedException
* if interrupted while waiting
*/
T get(long time, TimeUnit unit) throws InterruptedException;
}BoundedBlockingPool的实现如下：
package com.test.pool;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
public final class BoundedBlockingPool
extends <AbstractPool>
implements <BlockingPool>
{
private int size;
private BlockingQueue objects;
private Validator validator;
private ObjectFactory objectFactory;
private ExecutorService executor =
Executors.newCachedThreadPool();
private volatile boolean shutdownCalled;
public BoundedBlockingPool(
int size,
Validator validator,
ObjectFactory objectFactory)
{
super();
this.objectFactory = objectFactory;
this.size = size;
this.validator = validator;
objects = new LinkedBlockingQueue (size);
initializeObjects();
shutdownCalled = false;
}
public T get(long timeOut, TimeUnit unit)
{
if(!shutdownCalled)
{
T t = null;
try
{
t = objects.poll(timeOut, unit);
return t;
}
catch(InterruptedException ie)
{
Thread.currentThread().interrupt();
}
return t;
}
throw new IllegalStateException(
'Object pool is already shutdown');
}
public T get()
{
if(!shutdownCalled)
{
T t = null;
try
{
t = objects.take();
}
catch(InterruptedException ie)
{
Thread.currentThread().interrupt();
}
return t;
}
throw new IllegalStateException(
'Object pool is already shutdown');
}
public void shutdown()
{
shutdownCalled = true;
executor.shutdownNow();
clearResources();
}
private void clearResources()
{
for(T t : objects)
{
validator.invalidate(t);
}
}
@Override
protected void returnToPool(T t)
{
if(validator.isValid(t))
{
executor.submit(new ObjectReturner(objects, t));
}
}
@Override
protected void handleInvalidReturn(T t)
{
}
@Override
protected boolean isValid(T t)
{
return validator.isValid(t);
}
private void initializeObjects()
{
for(int i = 0; i < size; i++)
{
objects.add(objectFactory.createNew());
}
}
private class ObjectReturner
implements <Callable>
{
private BlockingQueue queue;
private E e;
public ObjectReturner(BlockingQueue queue, E e)
{
this.queue = queue;
this.e = e;
}
public Void call()
{
while(true)
{
try
{
queue.put(e);
break;
}
catch(InterruptedException ie)
{
Thread.currentThread().interrupt();
}
}
return null;
}
}
}上面是一个非常基本的对象池，它内部是基于一个LinkedBlockingQueue来实现的。这里唯一比较有意思的方法就是returnToPool。因为内部的存储是一个LinkedBlockingQueue实现的，如果我们直接把返回的对象扔进去的话，如果队列已满可能会阻塞住客户端。不过我们不希望客户端因为把对象放回池里这么个普通的方法就阻塞住了。所以我们把最终将对象插入到队列里的任务作为一个异步的的任务提交给一个Executor来执行，以便让客户端线程能立即返回。
现在我们将在自己的代码中使用上面这个对象池，用它来缓存数据库连接。我们需要一个校验器来验证数据库连接是否有效。
下面是这个JDBCConnectionValidator：
package com.test;
import java.sql.Connection;
import java.sql.SQLException;
import com.test.pool.Pool.Validator;
public final class JDBCConnectionValidator implements Validator < Connection >
{
public boolean isValid(Connection con)
{
if(con == null)
{
return false;
}
try
{
return !con.isClosed();
}
catch(SQLException se)
{
return false;
}
}
public void invalidate(Connection con)
{
try
{
con.close();
}
catch(SQLException se)
{
}
}
}还有一个JDBCObjectFactory，它将用来生成新的数据库连接对象：
package com.test;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import com.test.pool.ObjectFactory;
public class JDBCConnectionFactory implements ObjectFactory < Connection >
{
private String connectionURL;
private String userName;
private String password;
public JDBCConnectionFactory(
String driver,
String connectionURL,
String userName,
String password) {
super();
try
{
Class.forName(driver);
}
catch(ClassNotFoundException ce)
{
throw new IllegalArgumentException('Unable to find driver in classpath', ce);
}
this.connectionURL = connectionURL;
this.userName = userName;
this.password = password;
}
public Connection createNew()
{
try
{
return DriverManager.getConnection(
connectionURL,
userName,
password);
}
catch(SQLException se)
{
throw new IllegalArgumentException('Unable to create new connection', se);
}
}
}现在我们用上述的Validator和ObjectFactory来创建一个JDBC的连接池：
package com.test;
import java.sql.Connection;
import com.test.pool.Pool;
import com.test.pool.PoolFactory;
public class Main
{
public static void main(String[] args)
{
Pool < Connection > pool =
new BoundedBlockingPool < Connection > (
10,
new JDBCConnectionValidator(),
new JDBCConnectionFactory('', '', '', '')
);
//do whatever you like
}
}为了犒劳下能读完整篇文章的读者，我这再提供另一个非阻塞的对象池的实现，这个实现和前面的唯一不同就是即使对象不可用，它也不会让客户端阻塞，而是直接返回null。具体的实现在这：
package com.test.pool;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Semaphore;
public class BoundedPool < T > extends AbstractPool < T >
{
private int size;
private Queue < T > objects;
private Validator < T > validator;
private ObjectFactory < T > objectFactory;
private Semaphore permits;
private volatile boolean shutdownCalled;
public BoundedPool(
int size,
Validator < T > validator,
ObjectFactory < T > objectFactory)
{
super();
this.objectFactory = objectFactory;
this.size = size;
this.validator = validator;
objects = new LinkedList < T >();
initializeObjects();
shutdownCalled = false;
}
@Override
public T get()
{
T t = null;
if(!shutdownCalled)
{
if(permits.tryAcquire())
{
t = objects.poll();
}
}
else
{
throw new IllegalStateException('Object pool already shutdown');
}
return t;
}
@Override
public void shutdown()
{
shutdownCalled = true;
clearResources();
}
private void clearResources()
{
for(T t : objects)
{
validator.invalidate(t);
}
}
@Override
protected void returnToPool(T t)
{
boolean added = objects.add(t);
if(added)
{
permits.release();
}
}
@Override
protected void handleInvalidReturn(T t)
{
}
@Override
protected boolean isValid(T t)
{
return validator.isValid(t);
}
private void initializeObjects()
{
for(int i = 0; i < size; i++)
{
objects.add(objectFactory.createNew());
}
}
}考虑到我们现在已经有两种实现，非常威武了，得让用户通过工厂用具体的名称来创建不同的对象池了。工厂来了:
package com.test.pool;
import com.test.pool.Pool.Validator;
/**
* Factory and utility methods for
* {@link Pool} and {@link BlockingPool} classes
* defined in this package.
* This class supports the following kinds of methods:
*
*
<ul>
*
<li> Method that creates and returns a default non-blocking
* implementation of the {@link Pool} interface.
* </li>
*
*
<li> Method that creates and returns a
* default implementation of
* the {@link BlockingPool} interface.
* </li>
*
</ul>
*
* @author Swaranga
*/
public final class PoolFactory
{
private PoolFactory()
{
}
/**
* Creates a and returns a new object pool,
* that is an implementation of the {@link BlockingPool},
* whose size is limited by
* the <tt> size </tt> parameter.
*
* @param size the number of objects in the pool.
* @param factory the factory to create new objects.
* @param validator the validator to
* validate the re-usability of returned objects.
*
* @return a blocking object pool
* bounded by <tt> size </tt>
*/
public static < T > Pool < T >
newBoundedBlockingPool(
int size,
ObjectFactory < T > factory,
Validator < T > validator)
{
return new BoundedBlockingPool < T > (
size,
validator,
factory);
}
/*
* Creates a and returns a new object pool,
* that is an implementation of the {@link Pool}
* whose size is limited
* by the <tt> size </tt> parameter.
*
* @param size the number of objects in the pool.
* @param factory the factory to create new objects.
* @param validator the validator to validate
* the re-usability of returned objects.
*
* @return an object pool bounded by <tt> size </tt>
*/
public static < T > Pool < T > newBoundedNonBlockingPool(
int size,
ObjectFactory < T > factory,
Validator < T > validator)
{
return new BoundedPool < T >(size, validator, factory);
}
}现在我们的客户端就能用一种可读性更强的方式来创建对象池了：
package com.test;
import java.sql.Connection;
import com.test.pool.Pool;
import com.test.pool.PoolFactory;
public class Main
{
public static void main(String[] args)
{
Pool < Connection > pool =
PoolFactory.newBoundedBlockingPool(
10,
new JDBCConnectionFactory('', '', '', ''),
new JDBCConnectionValidator());
//do whatever you like
}
}以上是使用Java實作一個通用並發物件池的程式碼分享的詳細內容。更多資訊請關注PHP中文網其他相關文章！
陳述：
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn