专栏首页菩提树下的杨过common-pool2 学习:thrift连接池的另一种实现

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后,再次从对象池中借对象时,会重新创建对象。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • gradle项目中profile的实现

    gradle中并没有直接类似maven中的profile支持,只能变通的用其它方法来处理,在打包不同环境的应用时,通常会遇到二类问题: 一、不同的环境依赖的ja...

    菩提树下的杨过
  • thrift:swift项目笔记

    先声明:此swift不是Apple公司的那个swift开发语言,而是facebook的另一个开源项目。 facebook的thrift IDL文件,如果默认用t...

    菩提树下的杨过
  • 64位centos 下编译 hadoop 2.6.0 源码

    64位os下为啥要编译hadoop就不解释了,百度一下就能知道原因,下面是步骤: 前提:编译源码所在的机器,必须能上网,否则建议不要尝试了 一. 下载必要的组件...

    菩提树下的杨过
  • 放假也要好好学习/工作呀

    课代表
  • 简述JVM基础(三):垃圾收集器与内存分配策略

    了解了JVM中的内存分配以及数据存储的情况,那么怎么分配的方式是最高效的,对于已经过期无用的内存占用我们怎么回收起来,这个活必须得有人来干。好在JVM自带有垃圾...

    open
  • 搭建高性能的私有 Composer 镜像服务

    本文使用「署名 4.0 国际 (CC BY 4.0)」许可协议,欢迎转载、或重新修改使用,但需要注明来源。 署名 4.0 国际 (CC BY 4.0)

    soulteary
  • 一张图看懂IoT发展及安全性现状

    IoT物联网究竟是怎样一种存在?又很多人仍在质疑其是否会成为未来的时候,MicKinsey的数据已经表明,到2025年,将真实世界和数字世界连接起来讲能够产生1...

    FB客服
  • 图解神经网络机器翻译原理:LSTM、seq2seq到Zero-Shot

    【新智元导读】这篇刊登在 blog.statsbot.co 上的博文,通过对 LSTM、BRNN、seq2seq、Zero-Shot Translation 和...

    新智元
  • Scalaz(6)- typeclass:Functor-just map

      Functor是范畴学(Category theory)里的概念。不过无须担心,我们在scala FP编程里并不需要先掌握范畴学知识的。在scalaz里,F...

    用户1150956
  • 仔细观察物联网2.0-为什么它是不可避免的

    根据瞻博网络的数据,到2021年,一旦有460亿台设备,传感器和执行器连接在一起,那么我们将会进入一个更大,更强大和更稳定的物联网(IoT)。那么IoT 2.0...

    首席架构师智库

扫码关注云+社区

领取腾讯云代金券