gRPC学习记录(六)--客户端连接池
标签（空格分隔）： javaWEB
对于客户端来说建立一个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再跑了.
Paste_Image.png
