Java を使用してユニバーサル同時オブジェクト プールを実装するためのコード共有

この記事では主に 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();
}

あらゆるオブジェクトをサポートするために、上記のインターフェイスは意図的にシンプルかつユニバーサルになるように設計されています。これは、プールからオブジェクトを取得/返すメソッドと、オブジェクトを解放できるようにプールを閉じるメカニズムを提供します。

それでは、このインターフェースを実装してみましょう。始める前に、理想的なリリース方法では、クライアントから返されたオブジェクトが再利用できるかどうかを最初にチェックする必要があることに言及する価値があります。存在する場合は、そのオブジェクトをプールに戻します。そうでない場合は、そのオブジェクトを破棄します。このプール インターフェイスのすべての実装がこのルールに従うことを願っています。特定の実装クラスを開始する前に、まず 抽象クラス を作成して、後続の実装がこの点に従うように制限します。私たちが実装した抽象クラスは 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);
}

上記のクラスでは、オブジェクトをプールに戻す前にオブジェクト プールがオブジェクトを検証する必要があるようにします。特定の実装は、独自の 動作をカスタマイズするために、これら 3 つのメソッドを実装する方法を自由に選択できます。これらは、オブジェクトが有効かどうかを判断する方法、無効な場合にどうするか (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 と同じです。したがって、プールのインターフェースは次のようになります:

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);
 }
}

準備はほぼ完了しました。始める前に、これがこのオブジェクト プールの最大の特徴です。それは「新しい物体を創造する能力」です。私たちのオブジェクト プールは汎用であるため、プールを満たす新しいオブジェクトを生成する方法を知る必要があります。この機能はオブジェクト プール自体に依存することはできません。新しいオブジェクトを作成する共通の方法が必要です。これは、「新しいオブジェクトの作成方法」メソッドが 1 つだけある 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(
                &#39;Object pool is already shutdown&#39;);
    }

    public T get()
    {
        if(!shutdownCalled)
        {
            T t = null;
            try
            {
                t = objects.take();
            }

            catch(InterruptedException ie)
            {
                Thread.currentThread().interrupt();
            }
            return t;
        }

        throw new IllegalStateException(
                &#39;Object pool is already shutdown&#39;);
    }

    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(&#39;Unable to find driver in classpath&#39;, 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(&#39;Unable to create new connection&#39;, 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(&#39;&#39;, &#39;&#39;, &#39;&#39;, &#39;&#39;)
        );
        //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(&#39;Object pool already shutdown&#39;);
         }
         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(&#39;&#39;, &#39;&#39;, &#39;&#39;, &#39;&#39;),
        new JDBCConnectionValidator());
        //do whatever you like
     }
}

以上がJava を使用してユニバーサル同時オブジェクト プールを実装するためのコード共有の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。