使用Jakarta Commons Pool对象池技术

1. 为什么使用对象池技术

创建新的对象并初始化,可能会消耗很多时间。在这种对象的初始化工作中如果依赖一些rpc远程调用来创建对象,例如通过socket或者http连接远程服务资源,最典型的就是数据库服务以及远程队列(Remote Queue),建立连接 -> 发送数据 -> 接收连接 -> 释放连接的过程无疑对于客服端来说相当繁重。在需要大量或者频繁生成这样的对象的时候,就可能会对性能造成一些不可忽略的影响。要解决这个问题在软件层面上可以使用对象池技术(Object Pooling),而Jakarta Commons Pool框架则是处理对象池化的有力外援。
 

2. 对象池技术解释

对象池的基本思路是:将用过的对象保存起来,等下一次需要这种对象的时候,再拿出来重复使用,从而在一定程度上减少频繁创建对象所造成的开销。用于充当保存对象的“容器”的对象,被称为“对象池”(Object Pool,或简称Pool)。
 
并非所有对象都适合拿来池化――因为维护对象池也要造成一定开销。对生成时开销不大的对象进行池化,反而可能会出现“维护对象池的开销”大于“生成新对象的开销”,从而使性能降低的情况。但是对于生成时开销可观的对象,池化技术就是提高性能的有效策略了。
 

3. Jakarta Commons Pool对象池框架

在该框架中,主要工作有两类对象:
PoolableObjectFactory:用于管理被池化的对象的产生、激活、挂起、校验和销毁;
ObjectPool:用于管理要被池化的对象的借出和归还,并通知PoolableObjectFactory完成相应的工作;
 
相应地,使用Pool框架的过程,也就划分成“创立PoolableObjectFactory”、“使用ObjectPool”两种动作。
 

3.1 使用PoolableObjectFactory

Pool框架利用PoolableObjectFactory来管理被池化的对象。ObjectPool的实例在需要处理被池化的对象的产生、激活、挂起、校验和销毁工作时,就会调用跟它关联在一起的PoolableObjectFactory实例的相应方法来操作。
 
PoolableObjectFactory是在org.apache.commons.pool包中定义的一个接口。实际使用的时候需要利用这个接口的一个具体实现。Pool框架本身没有包含任何一种PoolableObjectFactory实现,需要根据情况自行创立。
 
创立PoolableObjectFactory的大体步骤是:
 
创建一个实现了PoolableObjectFactory接口的类。
import org.apache.commons.pool.PoolableObjectFactory;
public class PoolableObjectFactorySample
        implements PoolableObjectFactory {
    private static int counter = 0;
}
 
为这个类添加一个Object makeObject()方法。这个方法用于在必要时产生新的对象。
public Object makeObject() throws Exception {
    Object obj = String.valueOf(counter++);
    System.err.println("Making Object " + obj);
    return obj;
}
 
为这个类添加一个void activateObject(Object obj)方法。这个方法用于将对象“激活”――设置为适合开始使用的状态。
public void activateObject(Object obj) throws Exception {
    System.err.println("Activating Object " + obj);
}
 
为这个类添加一个void passivateObject(Object obj)方法。这个方法用于将对象“挂起”――设置为适合开始休眠的状态。
public void passivateObject(Object obj) throws Exception {
    System.err.println("Passivating Object " + obj);
}
 
为这个类添加一个boolean validateObject(Object obj)方法。这个方法用于校验一个具体的对象是否仍然有效,已失效的对象会被自动交给destroyObject方法销毁
public boolean validateObject(Object obj) {
    boolean result = (Math.random() > 0.5);
    System.err.println("Validating Object "
            + obj + " : " + result);
    return result;
}
 
为这个类添加一个void destroyObject(Object obj)方法。这个方法用于销毁被validateObject判定为已失效的对象。
public void destroyObject(Object obj) throws Exception {
    System.err.println("Destroying Object " + obj);
}
 
最后完成的PoolableObjectFactory类似这个样子:
import org.apache.commons.pool.PoolableObjectFactory;
public class PoolableObjectFactorySample
        implements PoolableObjectFactory {
    private static int counter = 0;
    public Object makeObject() throws Exception {
        Object obj = String.valueOf(counter++);
        System.err.println("Making Object " + obj);
        return obj;
    }
    public void activateObject(Object obj) throws Exception {
        System.err.println("Activating Object " + obj);
    }
    public void passivateObject(Object obj) throws Exception {
        System.err.println("Passivating Object " + obj);
    }
    public boolean validateObject(Object obj) {
        /* 以1/2的概率将对象判定为失效 */
        boolean result = (Math.random() > 0.5);
        System.err.println("Validating Object "
                + obj + " : " + result);
        return result;
    }
    public void destroyObject(Object obj) throws Exception {
        System.err.println("Destroying Object " + obj);
    }
}
 

3.2 使用ObjectPool

有了合适的PoolableObjectFactory之后,便可以开始请出ObjectPool来与之同台演出了。
 
ObjectPool是在org.apache.commons.pool包中定义的一个接口,实际使用的时候也需要利用这个接口的一个具体实现。Pool框架本身包含了若干种现成的ObjectPool实现,可以直接利用。如果都不合用,也可以根据情况自行创建。具体的创建方法,可以参看Pool框架的文档和源码。
 
ObjectPool的使用方法类似这样:
 
生成一个要用的PoolableObjectFactory类的实例。
PoolableObjectFactory factory = new PoolableObjectFactorySample();
 
利用这个PoolableObjectFactory实例为参数,生成一个实现了ObjectPool接口的类(例如StackObjectPool)的实例,作为对象池。
ObjectPool pool = new StackObjectPool(factory);
 
需要从对象池中取出对象时,调用该对象池的Object borrowObject()方法。
Object obj = null;
obj = pool.borrowObject();
 
需要将对象放回对象池中时,调用该对象池的void returnObject(Object obj)方法。
pool.returnObject(obj);
 
当不再需要使用一个对象池时,调用该对象池的void close()方法,释放它所占据的资源。
pool.close();
 
这些操作都可能会抛出异常,需要另外处理。
 
比较完整的使用ObjectPool的全过程,可以参考这段代码:
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.PoolableObjectFactory;
import org.apache.commons.pool.impl.StackObjectPool;
public class ObjectPoolSample {
    public static void main(String[] args) {
        Object obj = null;
        PoolableObjectFactory factory 
                = new PoolableObjectFactorySample();
        ObjectPool pool = new StackObjectPool(factory);
        try {
            for(long i = 0; i < 100 ; i++) {
                System.out.println("== " + i + " ==");
                obj = pool.borrowObject();
                System.out.println(obj);
                pool.returnObject(obj);
            }
            obj = null;//明确地设为null,作为对象已归还的标志
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        finally {
            try{
                if (obj != null) {//避免将一个对象归还两次
                    pool.returnObject(obj);
                }
                pool.close();
            }
            catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}
 
综上,UML图如下:

3.3 线程安全问题

有时候可能要在多线程环境下使用Pool框架,这时候就会遇到和Pool框架的线程安全程度有关的问题。
 
因为ObjectPool和KeyedObjectPool都是在org.apache.commons.pool中定义的接口,而在接口中无法使用“synchronized”来修饰方法,所以,一个ObjectPool下的各个方法是否是同步方法,完全要看具体的实现。而且,单纯地使用了同步方法,也并不能使对象就此在多线程环境里高枕无忧。
 
就Pool框架中自带的几个ObjectPool的实现而言,它们都在一定程度上考虑了在多线程环境中使用的情况。不过还不能说它们是完全“线程安全”的。
 
例如,这段代码有些时候就会有一些奇怪的表现,最后输出的结果比预期的要大:
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.impl.StackObjectPool;
class UnsafePicker extends Thread {
    private ObjectPool pool;
    public UnsafePicker(ObjectPool op) {
        pool = op;
    }
    public void run() {
        Object obj = null;
        try {
        /* 似乎&hellip;&hellip; */
            if ( pool.getNumActive() < 5 ) {
                sleep((long) (Math.random() * 10));
                obj = pool.borrowObject();
            }
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class UnsafeMultiThreadPoolingSample {
    public static void main(String[] args) {
        ObjectPool pool = new StackObjectPool
                (new BasePoolableObjectFactorySample());
        Thread ts[] = new Thread[20];
        for (int j = 0; j < ts.length; j++) {
            ts[j] =  new UnsafePicker(pool);
            ts[j].start();
        }
        try {
            Thread.sleep(1000);
            /* 然而&hellip;&hellip; */
            System.out.println("NumActive:" + pool.getNumActive());
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}
要避免这种情况,就要进一步采取一些措施才行:
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.impl.StackObjectPool;
class SafePicker extends Thread {
    private ObjectPool pool;
    public SafePicker(ObjectPool op) {
        pool = op;
    }
    public void run() {
        Object obj = null;
        try {
            /* 略加处理 */
            synchronized (pool) {
                if ( pool.getNumActive() < 5 ) {
                    sleep((long) (Math.random() * 10));
                    obj = pool.borrowObject();
                }
            }
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class SafeMultiThreadPoolingSample {
    public static void main(String[] args) {
        ObjectPool pool = new StackObjectPool
                (new BasePoolableObjectFactorySample());
        Thread ts[] = new Thread[20];
        for (int j = 0; j < ts.length; j++) {
            ts[j] =  new SafePicker(pool);
            ts[j].start();
        }
        try {
            Thread.sleep(1000);
            System.out.println("NumActive:" + pool.getNumActive());
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 
基本上,可以说Pool框架是线程相容的。但是要在多线程环境中使用,还需要作一些特别的处理。
 
 

4. Jedis中线程池的实例

下面看一个实例,由于近期在研究Redis,所以需要找一个可靠的redis驱动,有很多开源项目,详见链接,Jedis便是其中历史较早的。相比于其他驱动,Jedis提供了一个JedisPool用于管理redis连接的池,其主要工作的包括Pool.java,JedisPool.java和JedisPoolConfig.java。
 
Pool.java封装了一个GenericObjectPool,负责Jedis连接的产生、校验和销毁。
package redis.clients.util;
 
import org.apache.commons.pool.PoolableObjectFactory;
import org.apache.commons.pool.impl.GenericObjectPool;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisException;
 
public abstract class Pool<T> {
    private final GenericObjectPool internalPool;
 
    public Pool(final GenericObjectPool.Config poolConfig,
            PoolableObjectFactory factory) {
        this.internalPool = new GenericObjectPool(factory, poolConfig);
    }
 
    @SuppressWarnings("unchecked")
    public T getResource() {
        try {
            return (T) internalPool.borrowObject();
        } catch (Exception e) {
            throw new JedisConnectionException(
                    "Could not get a resource from the pool", e);
        }
    }
 
    public void returnResourceObject(final Object resource) {
        try {
            internalPool.returnObject(resource);
        } catch (Exception e) {
            throw new JedisException(
                    "Could not return the resource to the pool", e);
        }
    }
 
    public void returnBrokenResource(final T resource) {
    	returnBrokenResourceObject(resource);
    }
 
    public void returnResource(final T resource) {
    	returnResourceObject(resource);
    }
 
    protected void returnBrokenResourceObject(final Object resource) {
        try {
            internalPool.invalidateObject(resource);
        } catch (Exception e) {
            throw new JedisException(
                    "Could not return the resource to the pool", e);
        }
    }
 
    public void destroy() {
        try {
            internalPool.close();
        } catch (Exception e) {
            throw new JedisException("Could not destroy the pool", e);
        }
    }
}
 
JedisPool.java继承了Pool.java,内部写了一个Inner Class – BasePoolableObjectFactory,用于新建JedisPool实例时传入线程池建立、销毁、验证连接的基本方法。
package redis.clients.jedis;
 
import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.commons.pool.impl.GenericObjectPool.Config;
 
import redis.clients.util.Pool;
 
public class JedisPool extends Pool<Jedis> {
 
    public JedisPool(final Config poolConfig, final String host) {
        this(poolConfig, host, Protocol.DEFAULT_PORT, Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
    }
 
    public JedisPool(String host, int port) {
        this(new Config(), host, port, Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
    }
 
    public JedisPool(final String host) {
        this(host, Protocol.DEFAULT_PORT);
    }
 
    public JedisPool(final Config poolConfig, final String host, int port,
            int timeout, final String password) {
        this(poolConfig, host, port, timeout, password, Protocol.DEFAULT_DATABASE);
    }
 
    public JedisPool(final Config poolConfig, final String host, final int port) {
        this(poolConfig, host, port, Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
    }
 
    public JedisPool(final Config poolConfig, final String host, final int port, final int timeout) {
        this(poolConfig, host, port, timeout, null, Protocol.DEFAULT_DATABASE);
    }
 
    public JedisPool(final Config poolConfig, final String host, int port, int timeout, final String password,
                     final int database) {
        super(poolConfig, new JedisFactory(host, port, timeout, password, database));
    }
 
 
    public void returnBrokenResource(final BinaryJedis resource) {
    	returnBrokenResourceObject(resource);
    }
 
    public void returnResource(final BinaryJedis resource) {
    	returnResourceObject(resource);
    }
 
    /**
     * PoolableObjectFactory custom impl.
     */
    private static class JedisFactory extends BasePoolableObjectFactory {
        private final String host;
        private final int port;
        private final int timeout;
        private final String password;
        private final int database;
 
        public JedisFactory(final String host, final int port,
                final int timeout, final String password, final int database) {
            super();
            this.host = host;
            this.port = port;
            this.timeout = timeout;
            this.password = password;
            this.database = database;
        }
 
        public Object makeObject() throws Exception {
            final Jedis jedis = new Jedis(this.host, this.port, this.timeout);
 
            jedis.connect();
            if (null != this.password) {
                jedis.auth(this.password);
            }
            if( database != 0 ) {
                jedis.select(database);
            }
 
            return jedis;
        }
 
        public void destroyObject(final Object obj) throws Exception {
            if (obj instanceof Jedis) {
                final Jedis jedis = (Jedis) obj;
                if (jedis.isConnected()) {
                    try {
                        try {
                            jedis.quit();
                        } catch (Exception e) {
                        }
                        jedis.disconnect();
                    } catch (Exception e) {
 
                    }
                }
            }
        }
 
        public boolean validateObject(final Object obj) {
            if (obj instanceof Jedis) {
                final Jedis jedis = (Jedis) obj;
                try {
                    return jedis.isConnected() && jedis.ping().equals("PONG");
                } catch (final Exception e) {
                    return false;
                }
            } else {
                return false;
            }
        }
    }
}
 
JedisPoolConfig继承了GenericObjectPool.Config,用于指定一些线程池初始化参数。
package redis.clients.jedis;
 
import org.apache.commons.pool.impl.GenericObjectPool.Config;
 
/**
 * Subclass of org.apache.commons.pool.impl.GenericObjectPool.Config that
 * includes getters/setters so it can be more easily configured by Spring and
 * other IoC frameworks.
 * 
 * Spring example:
 * 
 * <bean id="jedisConfig" class="redis.clients.jedis.JedisPoolConfig"> <property name="testWhileIdle" value="true"></property> </bean>
 * 
 * <bean id="jedisPool" class="redis.clients.jedis.JedisPool" destroy-method="destroy"> <constructor-arg ref="jedisConfig">  * <constructor-arg value="localhost"> <constructor-arg type="int" value="6379"> </constructor-arg></constructor-arg></constructor-arg></bean>
 * 
 * For information on parameters refer to:
 * 
 * http://commons.apache.org/pool/apidocs/org/apache/commons/pool/impl/
 * GenericObjectPool.html
 */
public class JedisPoolConfig extends Config {
    public JedisPoolConfig() {
        // defaults to make your life with connection pool easier :)
        setTestWhileIdle(true);
        setMinEvictableIdleTimeMillis(60000);
        setTimeBetweenEvictionRunsMillis(30000);
        setNumTestsPerEvictionRun(-1);
    }
 
    public int getMaxIdle() {
        return maxIdle;
    }
 
    public void setMaxIdle(int maxIdle) {
        this.maxIdle = maxIdle;
    }
 
    public int getMinIdle() {
        return minIdle;
    }
 
    public void setMinIdle(int minIdle) {
        this.minIdle = minIdle;
    }
 
    public int getMaxActive() {
        return maxActive;
    }
 
    public void setMaxActive(int maxActive) {
        this.maxActive = maxActive;
    }
 
    public long getMaxWait() {
        return maxWait;
    }
 
    public void setMaxWait(long maxWait) {
        this.maxWait = maxWait;
    }
 
    public byte getWhenExhaustedAction() {
        return whenExhaustedAction;
    }
 
    public void setWhenExhaustedAction(byte whenExhaustedAction) {
        this.whenExhaustedAction = whenExhaustedAction;
    }
 
    public boolean isTestOnBorrow() {
        return testOnBorrow;
    }
 
    public void setTestOnBorrow(boolean testOnBorrow) {
        this.testOnBorrow = testOnBorrow;
    }
 
    public boolean isTestOnReturn() {
        return testOnReturn;
    }
 
    public void setTestOnReturn(boolean testOnReturn) {
        this.testOnReturn = testOnReturn;
    }
 
    public boolean isTestWhileIdle() {
        return testWhileIdle;
    }
 
    public void setTestWhileIdle(boolean testWhileIdle) {
        this.testWhileIdle = testWhileIdle;
    }
 
    public long getTimeBetweenEvictionRunsMillis() {
        return timeBetweenEvictionRunsMillis;
    }
 
    public void setTimeBetweenEvictionRunsMillis(
            long timeBetweenEvictionRunsMillis) {
        this.timeBetweenEvictionRunsMillis = timeBetweenEvictionRunsMillis;
    }
 
    public int getNumTestsPerEvictionRun() {
        return numTestsPerEvictionRun;
    }
 
    public void setNumTestsPerEvictionRun(int numTestsPerEvictionRun) {
        this.numTestsPerEvictionRun = numTestsPerEvictionRun;
    }
 
    public long getMinEvictableIdleTimeMillis() {
        return minEvictableIdleTimeMillis;
    }
 
    public void setMinEvictableIdleTimeMillis(long minEvictableIdleTimeMillis) {
        this.minEvictableIdleTimeMillis = minEvictableIdleTimeMillis;
    }
 
    public long getSoftMinEvictableIdleTimeMillis() {
        return softMinEvictableIdleTimeMillis;
    }
 
    public void setSoftMinEvictableIdleTimeMillis(
            long softMinEvictableIdleTimeMillis) {
        this.softMinEvictableIdleTimeMillis = softMinEvictableIdleTimeMillis;
    }
 
}
 
 

 

1 Comment on this Post.

  1. leslin

    Xu
    技术又加码了,文章写得越来越好!

Leave a Comment.