gRPC Java 应用教程

相关参考资料：

官方示例代码：

https://github.com/grpc/grpc-java/tree/master/examples/src/main/java/io/grpc/examples/helloworld

博客：

https://www.cnblogs.com/resentment/p/6792029.html

http://dev.dafan.info/detail/387292?p=74

https://www.cnblogs.com/ghj1976/p/5484968.html

http://www.javashuo.com/article/p-rmgcgjda-ce.html

1、准备步骤

一、eclipse下可安装以下插件做为proto3的编辑器

二、新建存储protocol buffer文件的路径 src/main/proto/test.proto

syntax = "proto3";
package SpringJavaConfigExample;

option java_package="com.example.java.grpc";
option java_outer_classname="GrpcTestServiceProto";
option java_multiple_files=true;

service Greeter{

     rpc SayHello(HelloRequest) returns (HelloReply){}
}

message HelloRequest{
    string name = 1;
}

message HelloReply {
    string message = 1;
}

三、配置maven的pom.xml文件，此处参阅github上的官方说明，gradle项目可查阅官方说明。

<build>
		<finalName>SpringJavaConfigExample</finalName>
		<extensions>
			<extension>
				<groupId>kr.motd.maven</groupId>
				<artifactId>os-maven-plugin</artifactId>
				<version>1.5.0.Final</version>
			</extension>
		</extensions>
		<plugins>
			<plugin>
				<groupId>org.xolstice.maven.plugins</groupId>
				<artifactId>protobuf-maven-plugin</artifactId>
				<version>0.5.0</version>
				<configuration>					
                <protocArtifact>com.google.protobuf:protoc:3.4.0:exe:${os.detected.classifier}
                </protocArtifact>
				<pluginId>grpc-java</pluginId>
				<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.7.0:exe:${os.detected.classifier}
                </pluginArtifact>
				</configuration>
				<executions>
					<execution>
						<goals>
							<goal>compile</goal>
							<goal>compile-custom</goal>
						</goals>
					</execution>
				</executions>
			</plugin>
		</plugins>
</build>

四、编译proto文件。cmd进入pom.xml所在文件路径，执行mvn compile 命令。执行后可在项目目录 /target/generated-sources/protobuf目录下找到生成的Java文件，拷贝到项目对应的包目录中便可。

如图所示，框选出的部分为基于proto文件自动生成的Java代码。

2、代码编写

六、编写项目实例，gRPC能够编写流式类型与普通类型的客户端和服务端，流式的客户端和服务端能够传递或返回多个对象，本例仅介绍普通类型，即仅传递和返回单个对象。

Client端分为几种不一样的调用模式：

阻塞调用（blockingCall）

异步调用（asyncCall）

Future直接调用（futureCallDirect）

Future回调调用（furueCallCallback）

可是对应的Server端代码是同样的。

查阅自动生成的GreeterGrpc.java可知：

Server端代码

public class GrpcTestServer implements Runnable {

	private static final Logger log = LoggerFactory.getLogger(GrpcTestServer.class);

	private int port = 50051;

	private Server server;

	private void start() throws IOException {
		server = ServerBuilder.forPort(port).addService(new GreeterImpl()).build().start();
		log.info("Server started, listening on {}", port);
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				System.err.println("*** shutting down gRPC server since JVM is shutting down");
				GrpcTestServer.this.stop();
				System.err.println("*** server shut down");
			}
		});
	}

	private void stop() {
		if (server != null) {
			server.shutdown();
		}
	}

	private void blockUntilShutdown() throws InterruptedException {
		if (server != null) {
			server.awaitTermination();
		}
	}

	public void startServer() throws IOException, InterruptedException {
		final GrpcTestServer server = new GrpcTestServer();
		start();
		blockUntilShutdown();
	}

	private class GreeterImpl extends GreeterGrpc.GreeterImplBase {
		@Override
		public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
			HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
			responseObserver.onNext(reply);
			responseObserver.onCompleted();
		}
	}

	@Override
	public void run() {
		try {
			startServer();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

	}

}

Client端代码

阻塞调用（blockingCall）

public class GrpcTestBlockingClient {

	private static final Logger log = LoggerFactory.getLogger(GrpcTestBlockingClient.class);

	private final ManagedChannel channel;
	private final GreeterGrpc.GreeterBlockingStub blockingStub;

	/** Construct client connecting to HelloWorld server at {@code host:port}. */
	  public GrpcTestBlockingClient(String host, int port) {
	    this(ManagedChannelBuilder.forAddress(host, port)
	        .usePlaintext(true)
	        .build());
	  }

	/** Construct client for accessing RouteGuide server using the existing channel. */
	  GrpcTestBlockingClient(ManagedChannel channel) {
	    this.channel = channel;
	    blockingStub = GreeterGrpc.newBlockingStub(channel);
	  }

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

	/** Say hello to server. */
	public void greet(String name) {
		log.info("Will try to greet " + name + " ...");
		HelloRequest request = HelloRequest.newBuilder().setName(name).build();
		HelloReply response;
		try {
			response = blockingStub.sayHello(request);
		} catch (StatusRuntimeException e) {
			log.info("{} RPC failed:{}",Level.WARN,e.getStatus());
			return;
		}
		log.info("Greeting: " + response.getMessage());
	}

}

异步调用（asyncCall）

public class GrpcTestAsyncClient {

	private static final Logger log = LoggerFactory.getLogger(GrpcTestAsyncClient.class);
	private final ManagedChannel channel;
	private final GreeterGrpc.GreeterStub stub;

	final CountDownLatch latch = new CountDownLatch(1);

	public GrpcTestAsyncClient(String host, int port) {
		this(ManagedChannelBuilder.forAddress(host, port)
				.usePlaintext(true).build());
	}

	/**
	 * Construct client for accessing RouteGuide server using the existing channel.
	 */
	GrpcTestAsyncClient(ManagedChannel channel) {
		this.channel = channel;
		stub = GreeterGrpc.newStub(channel);
	}

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

	public void greet(String name) {
		log.info("Will try to greet {} ...",name);
		HelloRequest request = HelloRequest.newBuilder().setName(name).build();
		StreamObserver<HelloReply> stream = new StreamObserver<HelloReply>() {

			@Override
			public void onNext(HelloReply value) {
				log.info("Greeting:{}", value.getMessage());
				latch.countDown();
			}

			@Override
			public void onError(Throwable t) {
				log.info("error:{}",t.getMessage());
			}

			@Override
			public void onCompleted() {
				log.info("Completed!");
				try {
					shutdown();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}

		};

		try {
			stub.sayHello(request, stream);
		} catch (StatusRuntimeException e) {
			log.info("{} RPC failed:{}", Level.WARN, e.getStatus());
			return;
		}
	}
}

Future直接调用（futureCallDirect）

public class GrpcTestFutureClient {

	private static final Logger log = LoggerFactory.getLogger(GrpcTestFutureClient.class);
	private final ManagedChannel channel;
	private final GreeterGrpc.GreeterFutureStub futureStub;

	public GrpcTestFutureClient(String host, int port) {
		this(ManagedChannelBuilder.forAddress(host, port)
				.usePlaintext(true).build());
	}

	/**
	 * Construct client for accessing RouteGuide server using the existing channel.
	 */
	GrpcTestFutureClient(ManagedChannel channel) {
		this.channel = channel;
		futureStub = GreeterGrpc.newFutureStub(channel);
	}

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

	public void greet(String name) throws InterruptedException, ExecutionException {
		log.info("Will try to greet " + name + " ...");
		HelloRequest request = HelloRequest.newBuilder().setName(name).build();
		try {
			ListenableFuture<HelloReply> listener = futureStub.sayHello(request);
			//listener.addListener(listener, executor);
			HelloReply response = listener.get();
			log.info("response:{}",response.getMessage());
			
		} catch (StatusRuntimeException e) {
			log.info("{0} RPC failed:{1}", Level.WARN, e.getStatus());
			return;
		}
	}
}

Future回调调用（furueCallCallback）

待完善。