package com.mingluck.grpc;/** * Created by Darren on 2016/11/11. */import io.grpc.Server;import io.grpc.ServerBuilder;import io.grpc.stub.StreamObserver; import java.io.IOException; import java.util.concurrent.atomic.AtomicInteger; import java.util.logging.Logger; public class HelloWorldServer { private static final Logger logger = Logger.getLogger(HelloWorldServer.class.getName()); /* The port on which the server should run */ private int port = 50051; private Server server; private void start() throws IOException { server = ServerBuilder.forPort(port) .addService(GreeterGrpc.bindService(new GreeterImpl())) .build() .start(); logger.info("Server started, listening on " + port); Runtime.getRuntime().addShutdownHook(new Thread() { @Override public void run() { // Use stderr here since the logger may have been reset by its JVM shutdown hook. System.err.println("*** shutting down gRPC server since JVM is shutting down"); HelloWorldServer.this.stop(); System.err.println("*** server shut down"); } }); } private void stop() { if (server != null) { server.shutdown(); } } /** * Await termination on the main thread since the grpc library uses daemon threads. */ private void blockUntilShutdown() throws InterruptedException { if (server != null) { server.awaitTermination(); } } /** * Main launches the server from the command line. */ public static void main(String[] args) throws IOException, InterruptedException { final HelloWorldServer server = new HelloWorldServer(); server.start(); server.blockUntilShutdown(); } private class GreeterImpl implements GreeterGrpc.Greeter { /** 原子Integer */ public AtomicInteger count = new AtomicInteger(0); @Override public void sayHello(HelloRequest req, StreamObserver responseObserver) { System.out.println("call sayHello"); HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName() + req.getSex()).build(); responseObserver.onNext(reply); responseObserver.onCompleted(); System.out.println(count.incrementAndGet() + Thread.currentThread().getName()); } } }
注意sayHello方法的重写
接下来实现HelloWorldClient.java
package com.mingluck.grpc;/** * Created by Darren on 2016/11/11. */import io.grpc.ManagedChannel;import io.grpc.ManagedChannelBuilder;import io.grpc.StatusRuntimeException; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; /** * A simple client that requests a greeting from the {@link HelloWorldServer}. */ public class HelloWorldClient { private static final Logger logger = Logger.getLogger(HelloWorldClient.class.getName()); private final ManagedChannel channel; private final GreeterGrpc.GreeterBlockingStub blockingStub; /** Construct client connecting to HelloWorld server at {@code host:port}. */ public HelloWorldClient(String host, int port) { channel = ManagedChannelBuilder.forAddress(host, port) .usePlaintext(true) .build(); blockingStub = GreeterGrpc.newBlockingStub(channel); } public void shutdown() throws InterruptedException { channel.shutdown().awaitTermination(5, TimeUnit.SECONDS); } /** Say hello to server. */ public void greet(String name) { logger.info("Will try to greet " + name + " ..."); HelloRequest request = HelloRequest.newBuilder().setName(name).setSex(" 女").build(); HelloReply response; try { response = blockingStub.sayHello(request); } catch (StatusRuntimeException e) { logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus()); return; } logger.info("Greeting: " + response.getMessage()); } /** * Greet server. If provided, the first element of {@code args} is the name to use in the * greeting. */ public static void main(String[] args) throws Exception { HelloWorldClient client = new HelloWorldClient("localhost", 50051); try { String user = "world"; if (args.length > 0) { user = args[0]; } client.greet(user); } finally { client.shutdown(); } } }
分别运行服务端和客户端，会看到rpc调用成功。
为了防止客户端不断调用带来的开销(短连接)，下面的例子给出了连接池的方式，代码如下
package com.mingluck.grpc;import org.apache.commons.pool2.BasePooledObjectFactory;import org.apache.commons.pool2.PooledObject;import org.apache.commons.pool2.impl.DefaultPooledObject;import org.apache.commons.pool2.impl.GenericObjectPool; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; /** * Created by darren on 2016/11/14. */ public class HelloWorldClientFactory extends BasePooledObjectFactory { @Override public HelloWorldClient create() throws Exception { return new HelloWorldClient("localhost", 50051); } @Override public PooledObject wrap(HelloWorldClient client) { return new DefaultPooledObject (client); } @Override public void destroyObject(PooledObject p) throws Exception { HelloWorldClient client = p.getObject(); client.shutdown(); super.destroyObject(p); } public static void main(String[] args) throws Exception { /** 连接池的配置 */ GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig(); /** 下面的配置均为默认配置,默认配置的参数可以在BaseObjectPoolConfig中找到 */ poolConfig.setMaxTotal(8); // 池中的最大连接数 poolConfig.setMinIdle(0); // 最少的空闲连接数 poolConfig.setMaxIdle(8); // 最多的空闲连接数 poolConfig.setMaxWaitMillis(-1); // 当连接池资源耗尽时,调用者最大阻塞的时间,超时时抛出异常 单位:毫秒数 poolConfig.setLifo(true); // 连接池存放池化对象方式,true放在空闲队列最前面,false放在空闲队列最后 poolConfig.setMinEvictableIdleTimeMillis(1000L * 60L * 30L); // 连接空闲的最小时间,达到此值后空闲连接可能会被移除,默认即为30分钟 poolConfig.setBlockWhenExhausted(true); // 连接耗尽时是否阻塞,默认为true /** 连接池创建 */ GenericObjectPool objectPool = new GenericObjectPool (new HelloWorldClientFactory(), poolConfig); new Thread(makeTask(objectPool)).start(); new Thread(makeTask(objectPool)).start(); new Thread(makeTask(objectPool)).start(); new Thread(makeTask(objectPool)).start(); Thread.sleep(100000); } private static Runnable makeTask(GenericObjectPool objectPool){ return () -> { HelloWorldClient client = null; try { client = objectPool.borrowObject(); } catch (Exception e) { e.printStackTrace(); } try { String req = "world!"; client.greet(req); } finally { /** 将连接对象返回给连接池 */ objectPool.returnObject(client); } }; } }
看下面的test.proto的内容： 
接下来可以写代码了，先实现服务端程序，HelloWorldServer.java