gRPC Java 应用教程

摘要: 介绍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
https://www.cnblogs.com/boshen-hzb/p/6555221.html

一、准备步骤

1、eclipse下可安装如下插件作为proto3的编辑器



2、新建存储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;
}

3、配置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>

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



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

二、代码编写

6、编写项目实例，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）

待完善。