Java gRPC Service Definition与Stub/BlockingStub使用
引言
在当今的分布式系统开发中，服务之间的通信变得越来越复杂。传统的HTTP/REST接口虽然简单易用，但在处理高并发、低延迟的场景时显得力不从心。gRPC作为一种高性能的远程过程调用（RPC）框架，凭借其高效的协议、强大的流式传输能力和跨语言支持，逐渐成为了现代微服务架构中的首选方案。
Java作为企业级应用开发的主流语言之一，自然也少不了对gRPC的支持。本文将以轻松诙谐的方式，带你深入了解如何在Java中定义gRPC服务，并使用Stub和BlockingStub进行客户端调用。我们将通过具体的代码示例和表格，逐步解析每一个关键点，帮助你快速上手gRPC开发。无论你是初学者还是有一定经验的开发者，相信这篇文章都能为你带来新的启发和收获。
什么是gRPC？
gRPC是由Google开发的一个开源RPC框架，它基于HTTP/2协议，使用Protocol Buffers（简称Protobuf）作为序列化格式。gRPC的核心优势在于其高效的数据传输、双向流式通信能力以及对多种编程语言的支持。相比传统的RESTful API，gRPC在性能上有显著提升，尤其是在处理大规模数据传输和实时性要求较高的场景下表现尤为出色。
gRPC的主要特点
- 高效的二进制协议：gRPC使用Protobuf作为默认的消息格式，相比JSON等文本格式，Protobuf具有更小的体积和更快的解析速度。
- 双向流式通信：gRPC支持四种调用模式：单向请求-响应、服务器端流、客户端流和双向流，能够满足不同业务场景的需求。
- 跨语言支持：gRPC支持多种编程语言，包括Java、Python、Go、C++等，方便不同语言的系统之间进行通信。
- 内置负载均衡：gRPC提供了丰富的负载均衡策略，能够自动选择最优的服务节点，提高系统的可用性和扩展性。
Java gRPC的环境搭建
在开始编写gRPC服务之前，我们需要先搭建好开发环境。以下是详细的步骤：
-
安装JDK：确保你的机器上已经安装了JDK 8或更高版本。你可以通过命令
java -version来检查当前的JDK版本。 -
创建Maven项目：我们使用Maven来管理项目的依赖。在IDE中创建一个新的Maven项目，并在
pom.xml文件中添加以下依赖项：<dependencies> <!-- gRPC核心库 --> <dependency> <groupId>io.grpc</groupId> <artifactId>grpc-netty-shaded</artifactId> <version>1.42.1</version> </dependency> <dependency> <groupId>io.grpc</groupId> <artifactId>grpc-protobuf</artifactId> <version>1.42.1</version> </dependency> <dependency> <groupId>io.grpc</groupId> <artifactId>grpc-stub</artifactId> <version>1.42.1</version> </dependency> <!-- Protobuf编译器插件 --> <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId> <version>3.19.1</version> </dependency> </dependencies> <build> <extensions> <extension> <groupId>kr.motd.maven</groupId> <artifactId>os-maven-plugin</artifactId> <version>1.6.2</version> </extension> </extensions> <plugins> <plugin> <groupId>org.xolstice.maven.plugins</groupId> <artifactId>protobuf-maven-plugin</artifactId> <version>0.6.1</version> <configuration> <protocArtifact>com.google.protobuf:protoc:3.19.1:exe:${os.detected.classifier}</protocArtifact> <pluginId>grpc-java</pluginId> <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.42.1:exe:${os.detected.classifier}</pluginArtifact> </configuration> <executions> <execution> <goals> <goal>compile</goal> <goal>compile-custom</goal> </goals> </execution> </executions> </plugin> </plugins> </build> -
安装Protobuf编译器：Protobuf编译器用于将
.proto文件编译成Java类。你可以通过以下命令安装Protobuf编译器：brew install protobuf # macOS sudo apt-get install protobuf-compiler # Ubuntu choco install protoc # Windows -
创建.proto文件：在
src/main/proto目录下创建一个名为hello.proto的文件，内容如下：syntax = "proto3"; option java_package = "com.example.grpc"; option java_outer_classname = "HelloWorldProto"; // 定义服务 service Greeter { rpc SayHello (HelloRequest) returns (HelloReply); } // 请求消息 message HelloRequest { string name = 1; } // 响应消息 message HelloReply { string message = 1; } -
编译.proto文件：在命令行中运行以下命令，将
.proto文件编译成Java类：mvn compile编译完成后，你会在
target/generated-sources/protobuf/java目录下看到生成的Java类文件。
定义gRPC服务
接下来，我们将在Java中实现gRPC服务。根据hello.proto文件中定义的服务接口，我们需要创建一个GreeterImpl类来实现GreeterGrpc.GreeterImplBase接口。
实现服务端
package com.example.grpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
public class HelloWorldServer {
private Server server;
public static void main(String[] args) throws Exception {
HelloWorldServer server = new HelloWorldServer();
server.start();
server.blockUntilShutdown();
}
private void start() throws Exception {
int port = 50051;
server = ServerBuilder.forPort(port)
.addService(new GreeterImpl())
.build()
.start();
System.out.println("Server started, listening on " + port);
}
private void blockUntilShutdown() throws InterruptedException {
if (server != null) {
server.awaitTermination();
}
}
static class GreeterImpl extends GreeterGrpc.GreeterImplBase {
@Override
public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
String name = request.getName();
String message = "Hello, " + name + "!";
HelloReply reply = HelloReply.newBuilder().setMessage(message).build();
responseObserver.onNext(reply);
responseObserver.onCompleted();
}
}
}在这个例子中，我们创建了一个简单的gRPC服务端，监听在50051端口上。GreeterImpl类实现了GreeterGrpc.GreeterImplBase接口中的sayHello方法，该方法接收一个HelloRequest对象并返回一个HelloReply对象。StreamObserver用于处理异步响应，onNext方法用于发送响应，onCompleted方法表示调用结束。
启动服务端
运行上述代码后，服务端会启动并监听指定的端口。你可以通过命令行工具如grpcurl来测试服务是否正常工作：
grpcurl -plaintext localhost:50051 com.example.grpc.Greeter/SayHello使用Stub进行客户端调用
现在我们已经有了一个gRPC服务端，接下来我们将编写客户端代码来调用这个服务。gRPC提供了两种类型的客户端：Stub和BlockingStub。Stub是异步的，而BlockingStub是同步的。根据不同的需求，我们可以选择适合的方式来调用服务。
异步调用（Stub）
Stub是一种异步的客户端调用方式，适用于需要非阻塞操作的场景。我们可以通过ManagedChannel来创建一个GreeterGrpc.GreeterStub对象，并使用StreamObserver来处理异步响应。
package com.example.grpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
public class HelloWorldClient {
private final ManagedChannel channel;
private final GreeterGrpc.GreeterStub asyncStub;
public HelloWorldClient(String host, int port) {
this.channel = ManagedChannelBuilder.forAddress(host, port)
.usePlaintext() // 禁用SSL/TLS
.build();
this.asyncStub = GreeterGrpc.newStub(channel);
}
public void shutdown() throws InterruptedException {
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
}
public void greet(String name) {
HelloRequest request = HelloRequest.newBuilder().setName(name).build();
asyncStub.sayHello(request, new StreamObserver<HelloReply>() {
@Override
public void onNext(HelloReply value) {
System.out.println("Received: " + value.getMessage());
}
@Override
public void onError(Throwable t) {
System.err.println("Error: " + t.getMessage());
}
@Override
public void onCompleted() {
System.out.println("Call completed.");
}
});
}
public static void main(String[] args) throws InterruptedException {
HelloWorldClient client = new HelloWorldClient("localhost", 50051);
try {
client.greet("World");
// 等待异步调用完成
Thread.sleep(1000);
} finally {
client.shutdown();
}
}
}在这个例子中，我们使用GreeterGrpc.newStub创建了一个异步的Stub对象。greet方法中，我们通过asyncStub.sayHello发起了一次异步调用，并传入了一个StreamObserver来处理响应。onNext方法会在接收到响应时被调用，onError方法用于处理错误，onCompleted方法表示调用结束。
同步调用（BlockingStub）
如果你希望使用同步的方式调用gRPC服务，可以使用BlockingStub。BlockingStub会阻塞当前线程，直到服务器返回响应。这种方式适用于简单的请求-响应场景，或者你不想处理复杂的异步逻辑。
package com.example.grpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
public class HelloWorldBlockingClient {
private final ManagedChannel channel;
private final GreeterGrpc.GreeterBlockingStub blockingStub;
public HelloWorldBlockingClient(String host, int port) {
this.channel = ManagedChannelBuilder.forAddress(host, port)
.usePlaintext() // 禁用SSL/TLS
.build();
this.blockingStub = GreeterGrpc.newBlockingStub(channel);
}
public void shutdown() throws InterruptedException {
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
}
public void greet(String name) {
HelloRequest request = HelloRequest.newBuilder().setName(name).build();
HelloReply response = blockingStub.sayHello(request);
System.out.println("Received: " + response.getMessage());
}
public static void main(String[] args) throws InterruptedException {
HelloWorldBlockingClient client = new HelloWorldBlockingClient("localhost", 50051);
try {
client.greet("World");
} finally {
client.shutdown();
}
}
}在这个例子中，我们使用GreeterGrpc.newBlockingStub创建了一个同步的BlockingStub对象。greet方法中，我们直接调用了blockingStub.sayHello，并等待服务器返回响应。由于是同步调用，因此不需要额外的回调函数来处理响应。
gRPC的四种调用模式
gRPC支持四种不同的调用模式，分别是：
-
Unary RPC：这是最常见的调用模式，客户端发送一个请求，服务器返回一个响应。我们在前面的例子中使用的正是这种模式。
-
Server Streaming RPC：客户端发送一个请求，服务器返回多个响应。适用于需要持续推送数据的场景，例如实时日志、股票行情等。
-
Client Streaming RPC：客户端发送多个请求，服务器返回一个响应。适用于需要上传大量数据的场景，例如文件上传、视频流等。
-
Bidirectional Streaming RPC：客户端和服务器都可以发送多个请求和响应。适用于需要双向通信的场景，例如聊天应用、实时协作编辑等。
Server Streaming RPC 示例
我们可以通过修改hello.proto文件来实现服务器端流式调用。首先，修改Greeter服务的定义，添加一个新的SayHelloStream方法：
service Greeter {
rpc SayHello (HelloRequest) returns (HelloReply);
rpc SayHelloStream (HelloRequest) returns (stream HelloReply);
}然后，修改服务端代码，实现SayHelloStream方法：
@Override
public void sayHelloStream(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
String name = request.getName();
for (int i = 0; i < 5; i++) {
String message = "Hello, " + name + "! This is message " + (i + 1);
HelloReply reply = HelloReply.newBuilder().setMessage(message).build();
responseObserver.onNext(reply);
try {
Thread.sleep(1000); // 模拟延迟
} catch (InterruptedException e) {
e.printStackTrace();
}
}
responseObserver.onCompleted();
}在客户端代码中，我们同样需要修改greet方法，以处理服务器端的流式响应：
public void greetStream(String name) {
HelloRequest request = HelloRequest.newBuilder().setName(name).build();
blockingStub.sayHelloStream(request, new StreamObserver<HelloReply>() {
@Override
public void onNext(HelloReply value) {
System.out.println("Received: " + value.getMessage());
}
@Override
public void onError(Throwable t) {
System.err.println("Error: " + t.getMessage());
}
@Override
public void onCompleted() {
System.out.println("Call completed.");
}
});
// 等待异步调用完成
Thread.sleep(6000);
}Client Streaming RPC 示例
为了实现客户端流式调用，我们需要在hello.proto文件中添加一个新的SayHelloClientStream方法：
service Greeter {
rpc SayHello (HelloRequest) returns (HelloReply);
rpc SayHelloStream (HelloRequest) returns (stream HelloReply);
rpc SayHelloClientStream (stream HelloRequest) returns (HelloReply);
}然后，修改服务端代码，实现SayHelloClientStream方法：
@Override
public StreamObserver<HelloRequest> sayHelloClientStream(StreamObserver<HelloReply> responseObserver) {
return new StreamObserver<HelloRequest>() {
List<String> names = new ArrayList<>();
@Override
public void onNext(HelloRequest value) {
names.add(value.getName());
}
@Override
public void onError(Throwable t) {
responseObserver.onError(t);
}
@Override
public void onCompleted() {
String message = "Hello, " + String.join(", ", names) + "! How are you all?";
HelloReply reply = HelloReply.newBuilder().setMessage(message).build();
responseObserver.onNext(reply);
responseObserver.onCompleted();
}
};
}在客户端代码中，我们可以通过asyncStub.sayHelloClientStream发起客户端流式调用，并使用StreamObserver来发送多个请求：
public void greetClientStream(List<String> names) {
StreamObserver<HelloRequest> requestObserver = asyncStub.sayHelloClientStream(new StreamObserver<HelloReply>() {
@Override
public void onNext(HelloReply value) {
System.out.println("Received: " + value.getMessage());
}
@Override
public void onError(Throwable t) {
System.err.println("Error: " + t.getMessage());
}
@Override
public void onCompleted() {
System.out.println("Call completed.");
}
});
for (String name : names) {
HelloRequest request = HelloRequest.newBuilder().setName(name).build();
requestObserver.onNext(request);
}
requestObserver.onCompleted();
// 等待异步调用完成
Thread.sleep(1000);
}Bidirectional Streaming RPC 示例
最后，我们来实现双向流式调用。在hello.proto文件中添加一个新的SayHelloBidirectionalStream方法：
service Greeter {
rpc SayHello (HelloRequest) returns (HelloReply);
rpc SayHelloStream (HelloRequest) returns (stream HelloReply);
rpc SayHelloClientStream (stream HelloRequest) returns (HelloReply);
rpc SayHelloBidirectionalStream (stream HelloRequest) returns (stream HelloReply);
}然后，修改服务端代码，实现SayHelloBidirectionalStream方法：
@Override
public StreamObserver<HelloRequest> sayHelloBidirectionalStream(StreamObserver<HelloReply> responseObserver) {
return new StreamObserver<HelloRequest>() {
@Override
public void onNext(HelloRequest value) {
String message = "Hello, " + value.getName() + "! I received your message.";
HelloReply reply = HelloReply.newBuilder().setMessage(message).build();
responseObserver.onNext(reply);
}
@Override
public void onError(Throwable t) {
responseObserver.onError(t);
}
@Override
public void onCompleted() {
responseObserver.onCompleted();
}
};
}在客户端代码中，我们可以通过asyncStub.sayHelloBidirectionalStream发起双向流式调用，并同时发送和接收消息：
public void greetBidirectionalStream(List<String> names) {
StreamObserver<HelloRequest> requestObserver = asyncStub.sayHelloBidirectionalStream(new StreamObserver<HelloReply>() {
@Override
public void onNext(HelloReply value) {
System.out.println("Received: " + value.getMessage());
}
@Override
public void onError(Throwable t) {
System.err.println("Error: " + t.getMessage());
}
@Override
public void onCompleted() {
System.out.println("Call completed.");
}
});
for (String name : names) {
HelloRequest request = HelloRequest.newBuilder().setName(name).build();
requestObserver.onNext(request);
try {
Thread.sleep(1000); // 模拟延迟
} catch (InterruptedException e) {
e.printStackTrace();
}
}
requestObserver.onCompleted();
// 等待异步调用完成
Thread.sleep(6000);
}总结
通过本文的学习，我们深入了解了如何在Java中定义gRPC服务，并使用Stub和BlockingStub进行客户端调用。我们不仅掌握了基本的Unary RPC调用，还学习了如何实现服务器端流、客户端流和双向流式调用。gRPC的强大之处在于它不仅提供了高效的通信机制，还支持多种调用模式，能够满足不同业务场景的需求。
在实际开发中，选择合适的调用模式非常重要。对于简单的请求-响应场景，BlockingStub是一个不错的选择；而对于需要处理大量数据或实时通信的场景，Stub和流式调用则更为合适。希望本文能为你在Java中使用gRPC提供一些有价值的参考和启发。
