基于Apache-Commons-Pool2实现Grpc客户端连接池

Posted 2020-12-24

概述

在项目运行过程中，有些操作对系统资源消耗较大，比如建立数据库连接、建立Redis连接等操作，我们希望一次性创建多个连接对象，并在以后需要使用时能直接使用已创建好的连接，达到提高性能的目的。池技术通过提前将一些占用较多资源的对象初始化，并将初始化后的对象保存到池中备用，达到提高应用服务性能的目的，数据库的JDBC连接池和Jedis连接池等都使用了池技术。
Apache-Commons-Pool2提供了一套池技术的规范接口和实现的通用逻辑，我们只需要实现其抽象出来的方法就可以了。这篇博文主要分享基于Apache-Commons-Pool2来实现Grpc连接池的应用。
关于Grpc相关的内容，大家如想了解基本的实现方法，可以参考我的另一篇博客（传送门）：http://blog.51cto.com/andrewli/2058908

核心组件

我们先来了解一下Apache-Commons-Pool2规范接口中涉及到的几个核心组件，包括：

• ObjectPool
  对象池，用于存储对象，并管理对象的入池和出池。对象池的实现类是 GenericObjectPool<T>；
• PoolConfig
  池属性，用于设置连接池的一些配置信息，比如最大池容量、超过池容量后的处理逻辑等。池属性的实现类是：GenericObjectPoolConfig；
• ObjectFactory
  对象工厂，在需要的时候生成新的对象实例，并放入池中。对象工厂的接口是：interface PooledObjectFactory<T>；
• ClientObject
  池对象，由对象工厂负责创建，并放入到对象池中；需要使用时从对象池中取出，执行对象的业务逻辑，使用完后再放回对象池。池对象的接口是：interface PooledObject<T>。

  核心组件依赖关系及其工作流程

  接口与类之间的依赖关系

  在梳理连接池相关的核心组件工作流程之前，我们先来了解一下核心组件涉及到的类和接口之间的继承和实现关系。

• 对象池类的继承关系
  对象池的最顶层接口是ObjectPool<T>，里面定义了对象池的基本方法，包括对象的添加、取出、校验、返还，以及获取处于Idle休眠状态的对象数量、获取处于Active状态的对象数量、清空池、关闭池。
  抽象类BaseGenericObjectPool<T>，定义了对象池的初始配置，并实现了对象池的基本接口方法。
  池类GenericObjectPool<T>继承了抽象类BaseGenericObjectPool<T>，并实现了ObjectPool<T>接口。其中添加了对象工厂、存储所有对象的Map、存储Idle对象的链式阻塞队列、当前已创建的对象数等属性。
  由于GenericObjectPool<T>类支持范型，我们要做的，就是指定GenericObjectPool<T>池类返回的池对象类型<T>，并设置对象工厂类、配置类等池属性；或者继承GenericObjectPool类以添加更多的自定义池特性。
  

• 池属性类的继承关系
  池属性的最上层接口是interface Cloneable，抽象类BaseObjectPoolConfig实现了这个接口，并定义了默认的池配置属性。
  GenericObjectPoolConfig类继承了BaseObjectPoolConfig，同样定义了默认的池配置属性值。
  我们可以直接使用GenericObjectPoolConfig类，或者继承GenericObjectPoolConfig类，根据自己的需求设置自定义池配置属性。
  

• 池内对象类的继承关系
  池内对象类实现了上层的PooledObject<T>接口，这个接口里面定义了一个池对象需要实现的各种方法。
  另外，池内对象类还需要定义类本身需要具备的成员属性和需要实现的业务方法。
  

• 对象工厂类的继承关系
  对象工厂类实现了最上层的PooledObjectFactory<T>接口，该接口定义了对象工厂的核心功能方法，包括：创建对象、销毁对象、校验对象、激活对象、钝化对象。
  

工作流程

根据上述对核心组件的类继承关系分析，我们可以梳理出一个流程，逐步实现各个组件，并组合成一套适用于我们业务的连接池架构。我们来看看这个流程该如何定义。
（1）定义我们的池内对象类 ClientObject，并结合我们的实际业务来实现上层接口的方法。
（2）定义对象工厂类ClientFactory，并结合我们的实际业务来实现上层接口的方法。
（3）定义池属性类ClientPoolConfig，结合我们的实际需求来设置属性值。
（4）使用对象池GenericObjectPool，指定泛型类型GenericObjectPool<ClientObject>。

连接池内部的核心业务逻辑：

池内对象的创建和返回逻辑是池技术里的关键，可以查看池对象的borrowObject方法去了解这部分细节内容。

应用实践

代码实现

根据上述对Apache-commons-pool2的特点和实现流程的分析，我们基于Grpc客户端连接池的应用场景，来进行代码实践，主要包括实现池内对象类 ClientObject和实现对象工厂类ClientFactory。
具体代码可以进入我的百度网盘下载，链接如下：
https://pan.baidu.com/s/1eaGpz6XN2a3ssw0eYsNLww

代码测试

为了验证我们的Grpc连接池的作用，我编写了一个测试方法，模拟以下场景，即开启10个线程，每个线程循环10次使用Grpc连接发送消息给grpc服务端，然后查看线程池中累计创建的连接对象个数、线程池中每个连接对象的被使用次数等信息。
通过测试输出的信息，我得到的结论是：不使用连接池时，总共需要进行100次Grpc连接并发送消息；使用连接池后，总共仅需要建立2次Grpc连接来发送100次消息，每个连接被调用了50次。
测试代码如下。

package com.cmcc.littlec.grpc.poolclient;
import com.cmcc.littlec.grpc.util.Constants;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

public class test {
    @SuppressWarnings("unchecked")
    public static GenericObjectPool<ClientObject> getClientPool(){
        ClientPoolFactory factory = new ClientPoolFactory(Constants.grpcHost, Constants.grpcPort);
        GenericObjectPoolConfig config = new GenericObjectPoolConfig();
        config.setMaxIdle(8);
        config.setMinIdle(3);
        config.setMaxTotal(18);
        config.setTestOnBorrow(true);
        config.setTestOnReturn(true);
        GenericObjectPool<ClientObject> clientPool = new GenericObjectPool<ClientObject>(factory, config);
        return clientPool;
    }

    public static void  main(String[] args ){
        final GenericObjectPool<ClientObject> clientPool = getClientPool();
        for(int i=0; i<10; i++) {
            Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        for (int i = 0; i < 10; i++) {
                            ClientObject client = clientPool.borrowObject();                                                        
                            String str = "hello, grpc client_" + i;//参数
                            try {
                                client.sayHello(str);
                            }catch(Exception e){
                                client.invalidate();
                            }
                            System.out.println("Thread : " + Thread.currentThread().getName() + "; clientPool size : " + clientPool.getCreatedCount());
                            System.out.println("clientObj : "+client.toString());
                            clientPool.returnObject(client);
                        }
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    }
                }
            });
            t.start();
            try {
                if(i%2==0) {
                    Thread.sleep(5000L);//每隔两个线程创建后停顿5S
                }
            }catch(Exception e){ }
        }
    }
}