common-pool2 學習：thrift連接池的另一種實現

對象池是一種很實用的技術，經典的例子就是數據庫連接池。去年曾經從零開始寫過一個thrift客戶端連接池。如果不想重造輪子，可以直接在apache開源項目commons-pool的基礎上開發。 

步驟：

一、定義對象工廠

package test.cn.mwee.service.paidui.pool;

import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

public class TProtocolFactory
        extends BasePooledObjectFactory<TProtocol> {


    private String host;
    private int port;
    private boolean keepAlive = true;

    public TProtocolFactory(String host, int port, boolean keepAlive) {
        this.host = host;
        this.port = port;
        this.keepAlive = keepAlive;
    }

    @Override
    public TProtocol create() throws TTransportException {
        TSocket tSocket = new TSocket(host, port);
        TTransport tTransport = new TFramedTransport(tSocket);
        tTransport.open();
        return new TCompactProtocol(tTransport);
    }


    @Override
    public PooledObject<TProtocol> wrap(TProtocol protocol) {
        return new DefaultPooledObject<>(protocol);
    }

    /**
     * 對象鈍化(即：從激活狀態轉入非激活狀態，returnObject時觸發）
     *
     * @param pooledObject
     * @throws TTransportException
     */
    @Override
    public void passivateObject(PooledObject<TProtocol> pooledObject) throws TTransportException {
        if (!keepAlive) {
            pooledObject.getObject().getTransport().flush();
            pooledObject.getObject().getTransport().close();
        }
    }


    /**
     * 對象激活(borrowObject時觸發）
     *
     * @param pooledObject
     * @throws TTransportException
     */
    @Override
    public void activateObject(PooledObject<TProtocol> pooledObject) throws TTransportException {
        if (!pooledObject.getObject().getTransport().isOpen()) {
            pooledObject.getObject().getTransport().open();
        }
    }


    /**
     * 對象銷毀(clear時會觸發）
     * @param pooledObject
     * @throws TTransportException
     */
    @Override
    public void destroyObject(PooledObject<TProtocol> pooledObject) throws TTransportException {
        passivateObject(pooledObject);
        pooledObject.markAbandoned();
    }


    /**
     * 驗證對象有效性
     *
     * @param p
     * @return
     */
    @Override
    public boolean validateObject(PooledObject<TProtocol> p) {
        if (p.getObject() != null) {
            if (p.getObject().getTransport().isOpen()) {
                return true;
            }
            try {
                p.getObject().getTransport().open();
                return true;
            } catch (TTransportException e) {
                e.printStackTrace();
            }
        }
        return false;
    }
}

有二個關鍵的方法，需要重寫：activateObject（對象激活） 及 passivateObject(對象鈍化)

 

二、定義對象池

package test.cn.mwee.service.paidui.pool;

import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

/**
 * Created by yangjunming on 6/7/16.
 */
public class AutoClearGenericObjectPool<T> extends GenericObjectPool<T> {

    public AutoClearGenericObjectPool(PooledObjectFactory<T> factory) {
        super(factory);
    }

    public AutoClearGenericObjectPool(PooledObjectFactory<T> factory, GenericObjectPoolConfig config) {
        super(factory, config);
    }

    @Override
    public void returnObject(T obj) {
        super.returnObject(obj);
        //空閑數>=激活數時，清理掉空閑連接
        if (getNumIdle() >= getNumActive()) {
            clear();
        }
    }

}

common-pools提供了對象池的默認實現：GenericObjectPool 但是該對象池中，對於處於空閑的對象，需要手動調用clear來釋放空閑對象，如果希望改變這一行為，可以自己派生自己的子類，重寫returnObject方法，上面的代碼中，每次歸還對象時，如果空閑的對象比激活的對象還要多（即：一半以上的對象都在打醬油），則調用clear方法。

 

三、使用示例：

package test.cn.mwee.service.paidui.pool;

import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.thrift.protocol.TProtocol;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * thrift 連接池測試
 */
public class ProtocolPoolTest {


    public static void main(String[] args) throws Exception {

        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(10);
        poolConfig.setMinIdle(1);
        poolConfig.setTestOnBorrow(true);

        ObjectPool<TProtocol> pool = new AutoClearGenericObjectPool<>(
                new TProtocolFactory("127.0.0.1", 13041, true), poolConfig);

        List<TProtocol> list = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            TProtocol protocol = pool.borrowObject();
            System.out.println(protocol.toString());
            if (i % 2 == 0) {
                //10個連接中，將偶數歸還
                pool.returnObject(protocol);
            } else {
                list.add(protocol);
            }
        }

        Random rnd = new Random();
        while (true) {
            System.out.println(String.format("active:%d,idea:%d", pool.getNumActive(), pool.getNumIdle()));
            Thread.sleep(5000);
            //每次還一個
            if (list.size() > 0) {
                int i = rnd.nextInt(list.size());
                pool.returnObject(list.get(i));
                list.remove(i);
            }

            //直到全部還完
            if (pool.getNumActive() <= 0) {
                break;
            }
        }

        System.out.println("------------------------");


        list.clear();
        //連接池為空，測試是否能重新創建新連接
        for (int i = 1; i <= 10; i++) {
            TProtocol protocol = pool.borrowObject();
            System.out.println(protocol.toString());
            if (i % 2 == 0) {
                pool.returnObject(protocol);
            } else {
                list.add(protocol);
            }
        }

        while (true) {
            System.out.println(String.format("active:%d,idea:%d", pool.getNumActive(), pool.getNumIdle()));
            Thread.sleep(5000);
            if (list.size() > 0) {
                int i = rnd.nextInt(list.size());
                pool.returnObject(list.get(i));
                list.remove(i);
            }

            if (pool.getNumActive() <= 0) {
                pool.close();
                break;
            }
        }

    }
}

注：需要從對象池取一個對象時，調用borrowObject（背后會調用activeObject激活對象），類似的，對象使用完之后，需要調用returnObject將對象放回對象池（背后會調用passivateObject使對象鈍化）

輸出：

org.apache.thrift.protocol.TCompactProtocol@146044d7
org.apache.thrift.protocol.TCompactProtocol@1e9e725a
org.apache.thrift.protocol.TCompactProtocol@516be40f
org.apache.thrift.protocol.TCompactProtocol@3c0a50da
org.apache.thrift.protocol.TCompactProtocol@3c0a50da
org.apache.thrift.protocol.TCompactProtocol@646be2c3
org.apache.thrift.protocol.TCompactProtocol@646be2c3
org.apache.thrift.protocol.TCompactProtocol@797badd3
org.apache.thrift.protocol.TCompactProtocol@797badd3
org.apache.thrift.protocol.TCompactProtocol@77be656f
active:5,idea:1
active:4,idea:2
active:3,idea:0
active:2,idea:1
active:1,idea:0
------------------------
org.apache.thrift.protocol.TCompactProtocol@221af3c0
org.apache.thrift.protocol.TCompactProtocol@62bd765
org.apache.thrift.protocol.TCompactProtocol@23a5fd2
org.apache.thrift.protocol.TCompactProtocol@78a2da20
org.apache.thrift.protocol.TCompactProtocol@78a2da20
org.apache.thrift.protocol.TCompactProtocol@dd3b207
org.apache.thrift.protocol.TCompactProtocol@dd3b207
org.apache.thrift.protocol.TCompactProtocol@551bdc27
org.apache.thrift.protocol.TCompactProtocol@551bdc27
org.apache.thrift.protocol.TCompactProtocol@58fdd99
active:5,idea:1
active:4,idea:2
active:3,idea:0
active:2,idea:1
active:1,idea:0

Process finished with exit code 0

從輸出上看，歸還對象后，再次取出時，並沒有創建新對象，而是直接使用了對象池中已經空閑的對象。當對象池中的所有對象都歸還變成空閑並被clear后，再次從對象池中借對象時，會重新創建對象。