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）

待完善。