logo
搜索
我的图书馆

发文章

发文工具

其他工具

留言交流
搜索
分享

一个通用并发对象池的实现 | 并发编程网

 aaie_ 2016-05-14


这篇文章里我们主要讨论下如何在Java里实现一个对象池。最近几年，Java虚拟机的性能在各方面都得到了极大的提升，因此对大多数对象而言，已经没有必要通过对象池来提高性能了。根本的原因是，创建一个新的对象的开销已经不像过去那样昂贵了。

然而，还是有些对象，它们的创建开销是非常大的，比如线程，数据库连接等这些非轻量级的对象。在任何一个应用程序里面，我们肯定会用到不止一个这样的对象。如果有一种很方便的创建管理这些对象的池，使得这些对象能够动态的重用，而客户端代码也不用关心它们的生命周期，还是会很给力的。

在真正开始写代码前，我们先来梳理下一个对象池需要完成哪些功能。

  • 如果有可用的对象，对象池应当能返回给客户端。
  • 客户端把对象放回池里后，可以对这些对象进行重用。
  • 对象池能够创建新的对象来满足客户端不断增长的需求。
  • 需要有一个正确关闭池的机制来确保关闭后不会发生内存泄露。

不用说了，上面几点就是我们要暴露给客户端的连接池的接口的基本功能。

我们的声明的接口如下：

01package com.test.pool;
02 
03/**
04 * Represents a cached pool of objects.
05 *
06 * @author Swaranga
07 *
08 * @param <T> the type of object to pool.
09 */
10public interface Pool<T>
11{
12 /**
13  * Returns an instance from the pool.
14  * The call may be a blocking one or a non-blocking one
15  * and that is determined by the internal implementation.
16  *
17  * If the call is a blocking call,
18  * the call returns immediately with a valid object
19  * if available, else the thread is made to wait
20  * until an object becomes available.
21  * In case of a blocking call,
22  * it is advised that clients react
23  * to {@link InterruptedException} which might be thrown
24  * when the thread waits for an object to become available.
25  *
26  * If the call is a non-blocking one,
27  * the call returns immediately irrespective of
28  * whether an object is available or not.
29  * If any object is available the call returns it
30  * else the call returns < code >null< /code >.
31  *
32  * The validity of the objects are determined using the
33  * {@link Validator} interface, such that
34  * an object < code >o< /code > is valid if
35  * < code > Validator.isValid(o) == true < /code >.
36  *
37  * @return T one of the pooled objects.
38  */
39 T get();
40 
41 /**
42  * Releases the object and puts it back to the pool.
43  *
44  * The mechanism of putting the object back to the pool is
45  * generally asynchronous,
46  * however future implementations might differ.
47  *
48  * @param t the object to return to the pool
49  */
50 
51 void release(T t);
52 
53 /**
54  * Shuts down the pool. In essence this call will not
55  * accept any more requests
56  * and will release all resources.
57  * Releasing resources are done
58  * via the < code >invalidate()< /code >
59  * method of the {@link Validator} interface.
60  */
61 
62 void shutdown();
63}

为了能够支持任意对象，上面这个接口故意设计得很简单通用。它提供了从池里获取/返回对象的方法，还有一个关闭池的机制，以便释放对象。

现在我们来实现一下这个接口。开始动手之前，值得一提的是，一个理想的release方法应该先尝试检查下这个客户端返回的对象是否还能重复使用。如果是的话再把它扔回池里，如果不是，就舍弃掉这个对象。我们希望这个Pool接口的所有实现都能遵循这个规则。在开始具体的实现类前，我们先创建一个抽象类，以便限制后续的实现能遵循这点。我们实现的抽象类就叫做AbstractPool，它的定义如下：

01package com.test.pool;
02 
03/**
04 * Represents an abstract pool, that defines the procedure
05 * of returning an object to the pool.
06 *
07 * @author Swaranga
08 *
09 * @param <T> the type of pooled objects.
10 */
11abstract class AbstractPool <T> implements Pool <T>
12{
13 /**
14  * Returns the object to the pool.
15  * The method first validates the object if it is
16  * re-usable and then puts returns it to the pool.
17  *
18  * If the object validation fails,
19  * some implementations
20  * will try to create a new one
21  * and put it into the pool; however
22  * this behaviour is subject to change
23  * from implementation to implementation
24  *
25  */
26 @Override
27 public final void release(T t)
28 {
29  if(isValid(t))
30  {
31   returnToPool(t);
32  }
33  else
34  {
35   handleInvalidReturn(t);
36  }
37 }
38 
39 protected abstract void handleInvalidReturn(T t);
40 
41 protected abstract void returnToPool(T t);
42 
43 protected abstract boolean isValid(T t);
44}

在上面这个类里，我们让对象池必须得先验证对象后才能把它放回到池里。具体的实现可以自由选择如何实现这三种方法，以便定制自己的行为。它们根据自己的逻辑来决定如何判断一个对象有效，无效的话应该怎么处理（handleInvalidReturn方法），怎么把一个有效的对象放回到池里（returnToPool方法）。

有了上面这几个类，我们就可以着手开始具体的实现了。不过还有个问题，由于上面这些类是设计成能支持通用的对象池的，因此具体的实现不知道该如何验证对象的有效性（因为对象都是泛型的）。因此我们还需要些别的东西来帮助我们完成这个。

我们需要一个通用的方法来完成对象的校验，而具体的实现不必关心对象是何种类型。因此我们引入了一个新的接口，Validator，它定义了验证对象的方法。这个接口的定义如下：

01package com.test.pool;
02 
03 /**
04  * Represents the functionality to
05  * validate an object of the pool
06  * and to subsequently perform cleanup activities.
07  *
08  * @author Swaranga
09  *
10  * @param < T > the type of objects to validate and cleanup.
11  */
12 public static interface Validator < T >
13 {
14  /**
15   * Checks whether the object is valid.
16   *
17   * @param t the object to check.
18   *
19   * @return <code>true</code>
20   * if the object is valid else <code>false</code>.
21   */
22  public boolean isValid(T t);
23 
24  /**
25   * Performs any cleanup activities
26   * before discarding the object.
27   * For example before discarding
28   * database connection objects,
29   * the pool will want to close the connections.
30   * This is done via the
31   * <code>invalidate()</code> method.
32   *
33   * @param t the object to cleanup
34   */
35 
36  public void invalidate(T t);
37 }

上面这个接口定义了一个检验对象的方法，以及一个把对象置为无效的方法。当准备废弃一个对象并清理内存的时候，invalidate方法就派上用场了。值得注意的是这个接口本身没有任何意义，只有当它在对象池里使用的时候才有意义，所以我们把这个接口定义到Pool接口里面。这和Java集合库里的Map和Map.Entry是一样的。所以我们的Pool接口就成了这样：

01package com.test.pool;
02 
03/**
04 * Represents a cached pool of objects.
05 *
06 * @author Swaranga
07 *
08 * @param < T > the type of object to pool.
09 */
10public interface Pool< T >
11{
12 /**
13  * Returns an instance from the pool.
14  * The call may be a blocking one or a non-blocking one
15  * and that is determined by the internal implementation.
16  *
17  * If the call is a blocking call,
18  * the call returns immediately with a valid object
19  * if available, else the thread is made to wait
20  * until an object becomes available.
21  * In case of a blocking call,
22  * it is advised that clients react
23  * to {@link InterruptedException} which might be thrown
24  * when the thread waits for an object to become available.
25  *
26  * If the call is a non-blocking one,
27  * the call returns immediately irrespective of
28  * whether an object is available or not.
29  * If any object is available the call returns it
30  * else the call returns < code >null< /code >.
31  *
32  * The validity of the objects are determined using the
33  * {@link Validator} interface, such that
34  * an object < code >o< /code > is valid if
35  * < code > Validator.isValid(o) == true < /code >.
36  *
37  * @return T one of the pooled objects.
38  */
39 T get();
40 
41 /**
42  * Releases the object and puts it back to the pool.
43  *
44  * The mechanism of putting the object back to the pool is
45  * generally asynchronous,
46  * however future implementations might differ.
47  *
48  * @param t the object to return to the pool
49  */
50 
51 void release(T t);
52 
53 /**
54  * Shuts down the pool. In essence this call will not
55  * accept any more requests
56  * and will release all resources.
57  * Releasing resources are done
58  * via the < code >invalidate()< /code >
59  * method of the {@link Validator} interface.
60  */
61 
62 void shutdown();
63 
64 /**
65  * Represents the functionality to
66  * validate an object of the pool
67  * and to subsequently perform cleanup activities.
68  *
69  * @author Swaranga
70  *
71  * @param < T > the type of objects to validate and cleanup.
72  */
73 public static interface Validator < T >
74 {
75  /**
76   * Checks whether the object is valid.
77   *
78   * @param t the object to check.
79   *
80   * @return <code>true</code>
81   * if the object is valid else <code>false</code>.
82   */
83  public boolean isValid(T t);
84 
85  /**
86   * Performs any cleanup activities
87   * before discarding the object.
88   * For example before discarding
89   * database connection objects,
90   * the pool will want to close the connections.
91   * This is done via the
92   * <code>invalidate()</code> method.
93   *
94   * @param t the object to cleanup
95   */
96 
97  public void invalidate(T t);
98 }
99}

准备工作已经差不多了，在最后开始前我们还需要一个终极武器，这才是这个对象池的杀手锏。就是“能够创建新的对象”。我们的对象池是泛型的，因此它们得知道如何去生成新的对象来填充这个池子。这个功能不能依赖于对象池本身，必须要有一个通用的方式来创建新的对象。通过一个ObjectFactory的接口就能完成这个，它只有一个“如何创建新的对象”的方法。我们的ObjectFactory接口如下：

01package com.test.pool;
02 
03/**
04 * Represents the mechanism to create
05 * new objects to be used in an object pool.
06 *
07 * @author Swaranga
08 *
09 * @param < T > the type of object to create.
10 */
11public interface ObjectFactory < T >
12{
13 /**
14  * Returns a new instance of an object of type T.
15  *
16  * @return T an new instance of the object of type T
17  */
18 public abstract T createNew();
19}

我们的工具类都已经搞定了，现在可以开始真正实现我们的Pool接口了。因为我们希望这个池能在并发程序里面使用，所以我们会创建一个阻塞的对象池，当没有对象可用的时候，让客户端先阻塞住。我们的阻塞机制是让客户端一直阻塞直到有对象可用为止。这样的话导致我们还需要再增加一个只阻塞一定时间的方法，如果在超时时间到来前有对象可用则返回，如果超时了就返回null而不是一直等待下去。这样的实现有点类似Java并发库里的LinkedBlockingQueue，因此真正实现前我们再暴露一个接口，BlockingPool，类似于Java并发库里的BlockingQueue接口。

这里是BlockingQueue的声明：

01package com.test.pool;
02 
03import java.util.concurrent.TimeUnit;
04 
05/**
06 * Represents a pool of objects that makes the
07 * requesting threads wait if no object is available.
08 *
09 * @author Swaranga
10 *
11 * @param < T > the type of objects to pool.
12 */
13public interface BlockingPool < T > extends Pool < T >
14{
15 /**
16  * Returns an instance of type T from the pool.
17  *
18  * The call is a blocking call,
19  * and client threads are made to wait
20  * indefinitely until an object is available.
21  * The call implements a fairness algorithm
22  * that ensures that a FCFS service is implemented.
23  *
24  * Clients are advised to react to InterruptedException.
25  * If the thread is interrupted while waiting
26  * for an object to become available,
27  * the current implementations
28  * sets the interrupted state of the thread
29  * to <code>true</code> and returns null.
30  * However this is subject to change
31  * from implementation to implementation.
32  *
33  * @return T an instance of the Object
34  * of type T from the pool.
35  */
36 T get();
37 
38 /**
39  * Returns an instance of type T from the pool,
40  * waiting up to the
41  * specified wait time if necessary
42  * for an object to become available..
43  *
44  * The call is a blocking call,
45  * and client threads are made to wait
46  * for time until an object is available
47  * or until the timeout occurs.
48  * The call implements a fairness algorithm
49  * that ensures that a FCFS service is implemented.
50  *
51  * Clients are advised to react to InterruptedException.
52  * If the thread is interrupted while waiting
53  * for an object to become available,
54  * the current implementations
55  * set the interrupted state of the thread
56  * to <code>true</code> and returns null.
57  * However this is subject to change
58  * from implementation to implementation.
59  *
60  *
61  * @param time amount of time to wait before giving up,
62  *   in units of <tt>unit</tt>
63  * @param unit a <tt>TimeUnit</tt> determining
64  *   how to interpret the
65  *        <tt>timeout</tt> parameter
66  *
67  * @return T an instance of the Object
68  * of type T from the pool.
69  *
70  * @throws InterruptedException
71  * if interrupted while waiting
72  */
73 
74 T get(long time, TimeUnit unit) throws InterruptedException;
75}

BoundedBlockingPool的实现如下：

001package com.test.pool;
002 
003import java.util.concurrent.BlockingQueue;
004import java.util.concurrent.Callable;
005import java.util.concurrent.ExecutorService;
006import java.util.concurrent.Executors;
007import java.util.concurrent.LinkedBlockingQueue;
008import java.util.concurrent.TimeUnit;
009public final class BoundedBlockingPool
010        extends <AbstractPool>
011        implements <BlockingPool>
012{
013    private int size;
014    private BlockingQueue  objects;
015    private Validator  validator;
016    private ObjectFactory  objectFactory;
017    private ExecutorService executor =
018            Executors.newCachedThreadPool();
019    private volatile boolean shutdownCalled;
020 
021    public BoundedBlockingPool(
022            int size,
023            Validator  validator,
024            ObjectFactory  objectFactory)
025    {
026        super();
027        this.objectFactory = objectFactory;
028        this.size = size;
029        this.validator = validator;
030        objects = new LinkedBlockingQueue (size);
031        initializeObjects();
032        shutdownCalled = false;
033    }
034 
035    public T get(long timeOut, TimeUnit unit)
036    {
037        if(!shutdownCalled)
038        {
039            T t = null;
040            try
041            {
042                t = objects.poll(timeOut, unit);
043                return t;
044            }
045            catch(InterruptedException ie)
046            {
047                Thread.currentThread().interrupt();
048            }
049            return t;
050        }
051        throw new IllegalStateException(
052                'Object pool is already shutdown');
053    }
054 
055    public T get()
056    {
057        if(!shutdownCalled)
058        {
059            T t = null;
060            try
061            {
062                t = objects.take();
063            }
064 
065            catch(InterruptedException ie)
066            {
067                Thread.currentThread().interrupt();
068            }
069            return t;
070        }
071 
072        throw new IllegalStateException(
073                'Object pool is already shutdown');
074    }
075 
076    public void shutdown()
077    {
078        shutdownCalled = true;
079        executor.shutdownNow();
080        clearResources();
081    }
082 
083    private void clearResources()
084    {
085        for(T t : objects)
086        {
087            validator.invalidate(t);
088        }
089    }
090 
091    @Override
092    protected void returnToPool(T t)
093    {
094        if(validator.isValid(t))
095        {
096            executor.submit(new ObjectReturner(objects, t));
097        }
098    }
099 
100    @Override
101    protected void handleInvalidReturn(T t)
102    {
103    }
104 
105    @Override
106    protected boolean isValid(T t)
107    {
108        return validator.isValid(t);
109    }
110 
111    private void initializeObjects()
112    {
113        for(int i = 0; i < size; i++)
114        {
115            objects.add(objectFactory.createNew());
116        }
117    }
118 
119    private class ObjectReturner
120            implements <Callable>
121    {
122        private BlockingQueue  queue;
123        private E e;
124        public ObjectReturner(BlockingQueue  queue, E e)
125        {
126            this.queue = queue;
127            this.e = e;
128        }
129 
130        public Void call()
131        {
132            while(true)
133            {
134                try
135                {
136                    queue.put(e);
137                    break;
138                }
139                catch(InterruptedException ie)
140                {
141                    Thread.currentThread().interrupt();
142                }
143            }
144            return null;
145        }
146 
147    }
148 
149}

上面是一个非常基本的对象池，它内部是基于一个LinkedBlockingQueue来实现的。这里唯一比较有意思的方法就是returnToPool。因为内部的存储是一个LinkedBlockingQueue实现的，如果我们直接把返回的对象扔进去的话，如果队列已满可能会阻塞住客户端。不过我们不希望客户端因为把对象放回池里这么个普通的方法就阻塞住了。所以我们把最终将对象插入到队列里的任务作为一个异步的的任务提交给一个Executor来执行，以便让客户端线程能立即返回。

现在我们将在自己的代码中使用上面这个对象池，用它来缓存数据库连接。我们需要一个校验器来验证数据库连接是否有效。

下面是这个JDBCConnectionValidator：

01package com.test;
02 
03import java.sql.Connection;
04import java.sql.SQLException;
05import com.test.pool.Pool.Validator;
06public final class JDBCConnectionValidator implements Validator < Connection >
07{
08    public boolean isValid(Connection con)
09    {
10        if(con == null)
11        {
12            return false;
13        }
14 
15        try
16        {
17            return !con.isClosed();
18        }
19        catch(SQLException se)
20        {
21            return false;
22        }
23 
24    }
25 
26    public void invalidate(Connection con)
27    {
28        try
29        {
30            con.close();
31        }
32        catch(SQLException se)
33        {
34        }
35    }
36 
37}

还有一个JDBCObjectFactory，它将用来生成新的数据库连接对象：

01package com.test;
02 
03import java.sql.Connection;
04import java.sql.DriverManager;
05import java.sql.SQLException;
06import com.test.pool.ObjectFactory;
07public class JDBCConnectionFactory implements ObjectFactory < Connection >
08{
09   private String connectionURL;
10   private String userName;
11   private String password;
12   public JDBCConnectionFactory(
13     String driver,
14     String connectionURL,
15     String userName,
16     String password) {
17     super();
18     try
19     {
20        Class.forName(driver);
21     }
22     catch(ClassNotFoundException ce)
23     {
24        throw new IllegalArgumentException('Unable to find driver in classpath', ce);
25     }
26     this.connectionURL = connectionURL;
27     this.userName = userName;
28     this.password = password;
29   }
30 
31   public Connection createNew()
32   {
33      try
34      {
35         return DriverManager.getConnection(
36            connectionURL,
37            userName,
38            password);
39      }
40      catch(SQLException se)
41      {
42         throw new IllegalArgumentException('Unable to create new connection', se);
43      }
44   }
45}

现在我们用上述的Validator和ObjectFactory来创建一个JDBC的连接池：

01package com.test;
02 
03import java.sql.Connection;
04import com.test.pool.Pool;
05import com.test.pool.PoolFactory;
06 
07public class Main
08{
09    public static void main(String[] args)
10    {
11        Pool < Connection > pool =
12            new BoundedBlockingPool < Connection > (
13            10,
14            new JDBCConnectionValidator(),
15            new JDBCConnectionFactory('', '', '', '')
16        );
17        //do whatever you like
18    }
19}

为了犒劳下能读完整篇文章的读者，我这再提供另一个非阻塞的对象池的实现，这个实现和前面的唯一不同就是即使对象不可用，它也不会让客户端阻塞，而是直接返回null。具体的实现在这：

01package com.test.pool;
02 
03import java.util.LinkedList;
04import java.util.Queue;
05import java.util.concurrent.Semaphore;
06 
07public class BoundedPool < T > extends AbstractPool < T >
08{
09    private int size;
10    private Queue < T > objects;
11    private Validator < T > validator;
12    private ObjectFactory < T > objectFactory;
13    private Semaphore permits;
14    private volatile boolean shutdownCalled;
15 
16    public BoundedPool(
17        int size,
18        Validator < T > validator,
19        ObjectFactory < T > objectFactory)
20        {
21        super();
22        this.objectFactory = objectFactory;
23        this.size = size;
24        this.validator = validator;
25        objects = new LinkedList < T >();
26        initializeObjects();
27        shutdownCalled = false;
28    }
29 
30    @Override
31    public T get()
32    {
33        T t = null;
34 
35        if(!shutdownCalled)
36        {
37            if(permits.tryAcquire())
38            {
39                t = objects.poll();
40            }
41 
42         }
43         else
44         {
45             throw new IllegalStateException('Object pool already shutdown');
46         }
47         return t;
48     }
49 
50     @Override
51     public void shutdown()
52     {
53         shutdownCalled = true;
54         clearResources();
55     }
56 
57     private void clearResources()
58     {
59         for(T t : objects)
60         {
61             validator.invalidate(t);
62         }
63     }
64 
65     @Override
66     protected void returnToPool(T t)
67     {
68         boolean added = objects.add(t);
69         if(added)
70         {
71             permits.release();
72         }
73     }
74 
75     @Override
76     protected void handleInvalidReturn(T t)
77     {
78     }
79 
80     @Override
81     protected boolean isValid(T t)
82     {
83         return validator.isValid(t);
84     }
85 
86     private void initializeObjects()
87     {
88         for(int i = 0; i < size; i++)
89         {
90             objects.add(objectFactory.createNew());
91         }
92     }
93}

考虑到我们现在已经有两种实现，非常威武了，得让用户通过工厂用具体的名称来创建不同的对象池了。工厂来了:

01package com.test.pool;
02 
03import com.test.pool.Pool.Validator;
04 
05/**
06 
07* Factory and utility methods for
08 
09* {@link Pool} and {@link BlockingPool} classes
10 
11* defined in this package.
12* This class supports the following kinds of methods:
13*
14*
15<ul>
16*
17<li> Method that creates and returns a default non-blocking
18*        implementation of the {@link Pool} interface.
19*   </li>
20*
21*
22<li> Method that creates and returns a
23*        default implementation of
24*        the {@link BlockingPool} interface.
25*   </li>
26*
27</ul>
28*
29* @author Swaranga
30*/
31public final class PoolFactory
32{
33    private PoolFactory()
34    {
35    }
36 
37/**
38* Creates a and returns a new object pool,
39* that is an implementation of the {@link BlockingPool},
40* whose size is limited by
41* the <tt> size </tt> parameter.
42*
43* @param size the number of objects in the pool.
44* @param factory the factory to create new objects.
45* @param validator the validator to
46* validate the re-usability of returned objects.
47*
48* @return a blocking object pool
49* bounded by <tt> size </tt>
50*/
51public static < T > Pool < T >
52newBoundedBlockingPool(
53int size,
54ObjectFactory < T > factory,
55Validator < T > validator)
56{
57    return new BoundedBlockingPool < T > (
58    size,
59    validator,
60    factory);
61}
62/*
63* Creates a and returns a new object pool,
64* that is an implementation of the {@link Pool}
65* whose size is limited
66* by the <tt> size </tt> parameter.
67*
68* @param size the number of objects in the pool.
69* @param factory the factory to create new objects.
70* @param validator the validator to validate
71* the re-usability of returned objects.
72*
73* @return an object pool bounded by <tt> size </tt>
74*/
75public static < T > Pool < T > newBoundedNonBlockingPool(
76    int size,
77    ObjectFactory < T > factory,
78    Validator < T > validator)
79{
80    return new BoundedPool < T >(size, validator, factory);
81}
82}

现在我们的客户端就能用一种可读性更强的方式来创建对象池了：

01package com.test;
02 
03import java.sql.Connection;
04import com.test.pool.Pool;
05import com.test.pool.PoolFactory;
06 
07public class Main
08{
09    public static void main(String[] args)
10    {
11        Pool < Connection > pool =
12        PoolFactory.newBoundedBlockingPool(
13        10,
14        new JDBCConnectionFactory('', '', '', ''),
15        new JDBCConnectionValidator());
16        //do whatever you like
17     }
18}

好吧，终于写完了，拖了这么久了。尽情使用和完善它吧，或者再多加几种实现。

    本站是提供个人知识管理的网络存储空间，所有内容均由用户发布，不代表本站观点。请注意甄别内容中的联系方式、诱导购买等信息，谨防诈骗。如发现有害或侵权内容，请点击一键举报。
    转藏 分享 献花（0）

    来自： aaie_ > 《JAVA》

    举报/认领

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多