Code-Sharing zur Implementierung eines universellen Pools gleichzeitiger Objekte mithilfe von Java

In diesem Artikel diskutieren wir hauptsächlich, wie man einen ObjektPool in Java implementiert. In den letzten Jahren wurde die Leistung der Java Virtual Machine in allen Aspekten erheblich verbessert, sodass für die meisten Objekte keine Leistungsverbesserung durch Objektpooling mehr erforderlich ist. Der Hauptgrund dafür ist, dass die Erstellung eines neuen Objekts nicht mehr so ​​​​kostspielig ist wie früher.

Es gibt jedoch immer noch einige Objekte, deren Erstellungsaufwand sehr hoch ist, wie zum Beispiel Threads, Datenbankverbindungen und andere nicht leichte Objekte. In jeder Anwendung werden wir auf jeden Fall mehr als ein solches Objekt verwenden. Wenn es eine bequeme Möglichkeit gibt, einen Pool dieser Objekte zu erstellen und zu verwalten, sodass diese Objekte dynamisch wiederverwendet werden können und sich der Client-Code nicht um ihren Lebenszyklus kümmern muss, ist dies immer noch der Fall mächtig.

Bevor wir tatsächlich mit dem Schreiben von Code beginnen, klären wir, welche Funktionen der nächste Objektpool erfüllen muss.

• Der Objektpool sollte zum Client zurückkehren können, wenn Objekte verfügbar sind.

• Nachdem der Kunde die Objekte wieder in den Pool gelegt hat, kann er diese Objekte wiederverwenden.

• Der Objektpool kann neue Objekte erstellen, um den wachsenden Anforderungen des Kunden gerecht zu werden.

• Es muss ein Mechanismus zum ordnungsgemäßen Schließen des Pools vorhanden sein, um sicherzustellen, dass nach dem Schließen keine Speicherverluste auftreten.

Unnötig zu erwähnen, dass die oben genannten Punkte die Grundfunktionen der Schnittstelle des Verbindungspools sind, die wir dem Client zugänglich machen möchten.

Unsere deklarierte Schnittstelle lautet wie folgt:

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

Um jedes Objekt zu unterstützen, ist die obige Schnittstelle bewusst einfach und universell gestaltet. Es bietet Methoden zum Abrufen/Zurückgeben von Objekten aus dem Pool sowie einen Mechanismus zum Schließen des Pools, damit Objekte freigegeben werden können.

Jetzt implementieren wir diese Schnittstelle. Bevor wir beginnen, ist es erwähnenswert, dass eine ideale Freigabemethode zunächst versuchen sollte, zu prüfen, ob das vom Client zurückgegebene Objekt wiederverwendet werden kann. Wenn ja, werfen Sie es zurück in den Pool. Wenn nicht, entsorgen Sie den Gegenstand. Wir hoffen, dass alle Implementierungen dieser Pool-Schnittstelle dieser Regel folgen können. Bevor wir mit der spezifischen Implementierungsklasse beginnen, erstellen wir zunächst eine abstrakte Klasse, um nachfolgende Implementierungen so einzuschränken, dass sie diesem Punkt folgen. Die von uns implementierte abstrakte Klasse heißt AbstractPool und hat folgende Definition:

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

In der obigen Klasse veranlassen wir den Objektpool, das Objekt zu überprüfen, bevor er es wieder in den Pool einfügt. Spezifische Implementierungen können frei wählen, wie sie diese drei Methoden implementieren, um ihr eigenes Verhalten anzupassen. Sie entscheiden auf der Grundlage ihrer eigenen Logik, wie sie feststellen, ob ein Objekt gültig ist, was zu tun ist, wenn es ungültig ist (Methode „handleInvalidReturn“) und wie ein gültiges Objekt wieder in den Pool eingefügt wird (Methode „returnToPool“).

Mit den oben genannten Klassen können wir mit der konkreten Umsetzung beginnen. Es besteht jedoch immer noch ein Problem, da die oben genannten Klassen für die Unterstützung universeller Objektpools konzipiert sind. Die spezifische Implementierung weiß nicht, wie die Gültigkeit der Objekte überprüft werden soll (da die Objekte alle generisch sind). Wir brauchen also etwas anderes, das uns dabei hilft.

Wir benötigen eine allgemeine Methode, um die Objektüberprüfung abzuschließen, und die spezifische Implementierung muss sich nicht um den Objekttyp kümmern. Aus diesem Grund haben wir eine neue Schnittstelle namens Validator eingeführt, die Methoden zur Validierung von Objekten definiert. Die Definition dieser Schnittstelle lautet wie folgt:

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

Die obige Schnittstelle definiert eine Methode zum Überprüfen des Objekts und eine Methode zum Ungültigmachen des Objekts. Wenn Sie bereit sind, ein Objekt zu verwerfen und den Speicher zu bereinigen, ist die Methode invalidate praktisch. Es ist erwähnenswert, dass diese Schnittstelle selbst keine Bedeutung hat. Sie macht nur dann Sinn, wenn sie in einem Objektpool verwendet wird. Daher definieren wir diese Schnittstelle in der Pool-Schnittstelle. Dies ist dasselbe wie Map und Map.Entry in der Java-Sammlungsbibliothek. Unsere Pool-Schnittstelle sieht also so aus:

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

Die Vorbereitungen sind fast abgeschlossen. Bevor wir endlich beginnen, brauchen wir eine ultimative Waffe. Es ist „die Fähigkeit, neue Objekte zu erstellen“. Unser Objektpool ist generisch, daher müssen sie wissen, wie sie neue Objekte generieren, um den Pool zu füllen. Diese Funktionalität kann nicht auf dem Objektpool selbst basieren, es muss eine gemeinsame Methode zum Erstellen neuer Objekte vorhanden sein. Dies kann über eine ObjectFactory-Schnittstelle erreicht werden, die nur über eine Methode zum Erstellen eines neuen Objekts verfügt. Unsere ObjectFactory-Schnittstelle sieht wie folgt aus:

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

Das obige ist der detaillierte Inhalt vonCode-Sharing zur Implementierung eines universellen Pools gleichzeitiger Objekte mithilfe von Java. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!