增加线程池应用,对于客户端最好用上线程池

quding · quding · commit 0d0bc6e9b9fd · 2017-02-09T00:07:43.000+08:00
diff --git a/gRPC-Demo/pom.xml b/gRPC-Demo/pom.xml
@@ -29,6 +29,13 @@
             <artifactId>grpc-stub</artifactId>
             <version>${grpc.version}</version>
         </dependency>
+
+        <!--客户端连接池-->
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-pool2</artifactId>
+            <version>2.4.2</version>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/gRPC-Demo/src/main/java/cn/mrdear/grpc/common/WorkCallBack.java b/gRPC-Demo/src/main/java/cn/mrdear/grpc/common/WorkCallBack.java
@@ -0,0 +1,11 @@
+package cn.mrdear.grpc.common;
+
+/**
+ * 给模板类使用的接口
+ * @author Niu Li
+ * @since 17-2-8
+ */
+public interface WorkCallBack<S> {
+
+  void callback(S s);
+}
diff --git a/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldClientPool.java b/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldClientPool.java
@@ -0,0 +1,82 @@
+package cn.mrdear.grpc.hellopool;
+
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+
+import cn.mrdear.grpc.common.WorkCallBack;
+
+/**
+ * @author Niu Li
+ * @since 17-2-8
+ */
+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);
+  }
+
+  /**
+   * 从连接池获取对象
+   */
+  public static HelloWorldClientSingle borrowObject(){
+    try {
+      HelloWorldClientSingle clientSingle = objectPool.borrowObject();
+      System.out.println("总创建线程数"+objectPool.getCreatedCount());
+      return clientSingle;
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+    //连接池失败则主动创建
+    return createClient();
+  }
+
+  /**
+   * 还送对象
+   */
+  public static void returnObject(HelloWorldClientSingle clientSingle){
+    objectPool.returnObject(clientSingle);
+  }
+
+  /**
+   * 当连接池异常,则主动创建对象
+   */
+  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 {
+        /** 将连接对象返回给连接池 */
+        returnObject(client);
+      }
+    };
+  }
+
+}
diff --git a/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldClientSingle.java b/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldClientSingle.java
@@ -0,0 +1,44 @@
+package cn.mrdear.grpc.hellopool;
+
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.StatusRuntimeException;
+import io.grpc.examples.helloworld.GreeterGrpc;
+import io.grpc.examples.helloworld.HelloReply;
+import io.grpc.examples.helloworld.HelloRequest;
+
+/**
+ * @author Niu Li
+ * @date 2017/1/28
+ */
+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());
+    }
+}
diff --git a/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldClientTest.java b/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldClientTest.java
@@ -0,0 +1,19 @@
+package cn.mrdear.grpc.hellopool;
+
+/**
+ * @author Niu Li
+ * @since 17-2-8
+ */
+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();
+    }
+  }
+
+
+
+}
diff --git a/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldFactory.java b/gRPC-Demo/src/main/java/cn/mrdear/grpc/hellopool/HelloWorldFactory.java
@@ -0,0 +1,27 @@
+package cn.mrdear.grpc.hellopool;
+
+import org.apache.commons.pool2.BasePooledObjectFactory;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+
+/**
+ * 连接池所需工厂
+ * @author Niu Li
+ * @since 17-2-8
+ */
+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);
+  }
+}
