gRPC学习记录(六)–客户端连接池

作者:屈定

出处:https://www.jianshu.com/u/fd3d072db53b


对于客户端来说建立一个channel是昂贵的,因为创建channel需要连接,但是建立一个stub是很简单的,就像创建一个普通对象,因此Channel就需要复用,也就是说需要实现一个连接池应用.本文使用commons-pool2来实现连接池应用. 该案例可以当成连接池模板案例.

<!--客户端连接池-->
       <dependency>
           <groupId>org.apache.commons</groupId>
           <artifactId>commons-pool2</artifactId>
           <version>2.4.2</version>
       </dependency>

写法的封装类似我的另一篇博文: Redis学习记录(二)–使用Jedis连接,模仿SpringJdbcTemplate模板类写法.

首先定义一个回调接口.

public interface WorkCallBack<S> {

 void callback(S s);
}

再者定义一个产生连接池的工厂,需要继承自BasePooledObjectFactory,其用处是产生连接池中需要的客户端.

public class HelloWorldFactory extends BasePooledObjectFactory<HelloWorldClientSingle> {

 private String host = "127.0.0.1";

 private int port = 50051;

 @Override
 public HelloWorldClientSingle create() throws Exception {
   return new HelloWorldClientSingle(this.host,this.port);
 }

 @Override
 public PooledObject<HelloWorldClientSingle> wrap(HelloWorldClientSingle helloWorldClientSingle) {
   return new DefaultPooledObject<>(helloWorldClientSingle);
 }

 @Override
 public void destroyObject(PooledObject<HelloWorldClientSingle> p) throws Exception {
   p.getObject().shutdown();
   super.destroyObject(p);
 }
}

实现相应的连接池,注意看execute()方法,该方法是一个模板方法,进行从池中获取客户端和归还客户端给连接池,其主要逻辑都定义在WorkCallBack接口中,也因此从服务端拿回的数据要在其里面处理完毕.

public class HelloWorldClientPool {

 private static GenericObjectPool<HelloWorldClientSingle> objectPool = null;

 static {
   // 连接池的配置
   GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
   // 池中的最大连接数
   poolConfig.setMaxTotal(8);
   // 最少的空闲连接数
   poolConfig.setMinIdle(0);
   // 最多的空闲连接数
   poolConfig.setMaxIdle(8);
   // 当连接池资源耗尽时,调用者最大阻塞的时间,超时时抛出异常 单位:毫秒数
   poolConfig.setMaxWaitMillis(-1);
   // 连接池存放池化对象方式,true放在空闲队列最前面,false放在空闲队列最后
   poolConfig.setLifo(true);
   // 连接空闲的最小时间,达到此值后空闲连接可能会被移除,默认即为30分钟
   poolConfig.setMinEvictableIdleTimeMillis(1000L * 60L * 30L);
   // 连接耗尽时是否阻塞,默认为true
   poolConfig.setBlockWhenExhausted(true);
   // 连接池创建
   objectPool = new GenericObjectPool<>(new HelloWorldFactory(), poolConfig);
 }

 /**
  * 从连接池获取对象
  */
 private static HelloWorldClientSingle borrowObject(){
   try {
     HelloWorldClientSingle clientSingle = objectPool.borrowObject();
     System.out.println("总创建线程数"+objectPool.getCreatedCount());
     return clientSingle;
   } catch (Exception e) {
     e.printStackTrace();
   }
   //连接池失败则主动创建
   return createClient();
 }

 /**
  * 当连接池异常,则主动创建对象
  */
 private static HelloWorldClientSingle createClient(){
   return new HelloWorldClientSingle("127.0.0.1", 55001);
 }

 /**
  * 执行器
  * @param workCallBack 主要服务内容
  */
 public static Runnable execute(WorkCallBack<HelloWorldClientSingle> workCallBack){
   return () -> {
     HelloWorldClientSingle client = borrowObject();
     try {
      workCallBack.callback(client);
     } finally {
       /** 将连接对象返回给连接池 */
       objectPool.returnObject(client);
     }
   };
 }

}

对于客户端来说,因为存根是不能共享的,所以存根要放在调用函数中实例化,不要担心性能问题,这个操作就是创建一个对象.greeterBlockingStub = GreeterGrpc.newBlockingStub(channel).withCompression("gzip");该代码从之前的构造函数中移位到了greet中.

public class HelloWorldClientSingle {

   private final ManagedChannel channel; //一个gRPC信道
   private GreeterGrpc.GreeterBlockingStub greeterBlockingStub;//阻塞/同步 存根

  //初始化信道和存根
   public HelloWorldClientSingle(String host,int port){
     channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext(true).build();
   }

   public void shutdown() throws InterruptedException {
       channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
   }

   //客户端方法
   public  void greet(String name){
       //需要用到存根时创建,不可复用
       greeterBlockingStub = GreeterGrpc.newBlockingStub(channel).withCompression("gzip");
       HelloRequest request = HelloRequest.newBuilder().setName(name).build();
       HelloReply response;
       try {
           response = greeterBlockingStub.sayHello(request);
       } catch (StatusRuntimeException e) {
           System.out.println("RPC调用失败:"+e.getMessage());
           return;
       }
       System.out.println("服务器返回信息:"+response.getMessage());
   }
}

写一个简单的测试类

public class HelloWorldClientTest {

 public static void main(String[] args) {
   for (int i = 0; i < 100; i++) {
     new Thread( HelloWorldClientPool.execute(clientSingle -> {
         clientSingle.greet("world");
     })).start();
   }
 }
}

因为我限定的8个客户端,所以最高也就8个channel再跑了.

参考博文: http://zk-chs.iteye.com/blog/2308730
b
代码地址:
https://github.com/nl101531/JavaWEB

赞(0) 打赏

如未加特殊说明,此网站文章均为原创,转载必须注明出处。Java 技术驿站 » gRPC学习记录(六)–客户端连接池
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

关注【Java 技术驿站】公众号,每天早上 8:10 为你推送一篇技术文章

扫描二维码关注我!


关注【Java 技术驿站】公众号 回复 “VIP”,获取 VIP 地址永久关闭弹出窗口

免费获取资源

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏