本文目录导读：
我来详细介绍如何在Java中实现gRPC的完整案例。
环境准备
添加Maven依赖
<dependencies>
<!-- gRPC核心依赖 -->
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
<version>1.59.0</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<version>1.59.0</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<version>1.59.0</version>
</dependency>
<!-- Protocol Buffers -->
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.25.1</version>
</dependency>
<!-- 注解处理 -->
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
<version>1.3.2</version>
</dependency>
</dependencies>
<build>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.7.1</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.25.1:exe:${os.detected.classifier}</protocArtifact>
<pluginId>grpc-java</pluginId>
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.59.0:exe:${os.detected.classifier}</pluginArtifact>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
定义Protobuf服务
创建 user.proto 文件
syntax = "proto3";
option java_multiple_files = true;
option java_package = "com.example.grpc";
option java_outer_classname = "UserProto";
package user;
// 用户服务定义
service UserService {
// 简单RPC - 获取用户信息
rpc GetUser (UserRequest) returns (UserResponse);
// 服务器端流式RPC - 获取用户列表
rpc ListUsers (UserListRequest) returns (stream UserResponse);
// 客户端流式RPC - 批量创建用户
rpc CreateUsers (stream CreateUserRequest) returns (CreateUsersResponse);
// 双向流式RPC - 用户聊天
rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}
// 用户请求
message UserRequest {
int32 user_id = 1;
}
// 用户列表请求
message UserListRequest {
int32 page = 1;
int32 page_size = 2;
}
// 创建用户请求
message CreateUserRequest {
string name = 1;
int32 age = 2;
string email = 3;
}
// 用户响应
message UserResponse {
int32 user_id = 1;
string name = 2;
int32 age = 3;
string email = 4;
string created_at = 5;
}
// 批量创建用户响应
message CreateUsersResponse {
int32 created_count = 1;
string message = 2;
}
// 聊天消息
message ChatMessage {
string user_name = 1;
string message = 2;
int64 timestamp = 3;
}
编译生成代码
# 使用Maven编译生成Java代码 mvn clean compile
实现服务器端
UserServiceImpl.java
package com.example.grpc.server;
import com.example.grpc.*;
import io.grpc.stub.StreamObserver;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.logging.Logger;
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {
private static final Logger logger = Logger.getLogger(UserServiceImpl.class.getName());
private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
// 简单RPC - 获取单个用户
@Override
public void getUser(UserRequest request, StreamObserver<UserResponse> responseObserver) {
logger.info("Received getUser request for user ID: " + request.getUserId());
// 模拟从数据库获取用户数据
UserResponse response = UserResponse.newBuilder()
.setUserId(request.getUserId())
.setName("张三")
.setAge(25)
.setEmail("zhangsan@example.com")
.setCreatedAt(LocalDateTime.now().format(formatter))
.build();
responseObserver.onNext(response);
responseObserver.onCompleted();
}
// 服务器端流式RPC - 获取用户列表
@Override
public void listUsers(UserListRequest request, StreamObserver<UserResponse> responseObserver) {
logger.info("Received listUsers request - page: " + request.getPage() + ", size: " + request.getPageSize());
// 模拟发送用户列表
for (int i = 1; i <= request.getPageSize(); i++) {
UserResponse user = UserResponse.newBuilder()
.setUserId(i)
.setName("用户" + i)
.setAge(20 + i)
.setEmail("user" + i + "@example.com")
.setCreatedAt(LocalDateTime.now().format(formatter))
.build();
responseObserver.onNext(user);
try {
Thread.sleep(100); // 模拟延迟
} catch (InterruptedException e) {
e.printStackTrace();
}
}
responseObserver.onCompleted();
}
// 客户端流式RPC - 批量创建用户
@Override
public StreamObserver<CreateUserRequest> createUsers(StreamObserver<CreateUsersResponse> responseObserver) {
logger.info("Received createUsers request");
return new StreamObserver<CreateUserRequest>() {
int count = 0;
@Override
public void onNext(CreateUserRequest request) {
count++;
logger.info("Creating user: " + request.getName());
// 模拟创建用户操作
}
@Override
public void onError(Throwable t) {
logger.severe("Error in createUsers: " + t.getMessage());
}
@Override
public void onCompleted() {
logger.info("Completed creating " + count + " users");
CreateUsersResponse response = CreateUsersResponse.newBuilder()
.setCreatedCount(count)
.setMessage("成功创建了 " + count + " 个用户")
.build();
responseObserver.onNext(response);
responseObserver.onCompleted();
}
};
}
// 双向流式RPC - 聊天
@Override
public StreamObserver<ChatMessage> chat(StreamObserver<ChatMessage> responseObserver) {
logger.info("New chat session started");
return new StreamObserver<ChatMessage>() {
@Override
public void onNext(ChatMessage request) {
logger.info("Received from " + request.getUserName() + ": " + request.getMessage());
// 自动回复
ChatMessage reply = ChatMessage.newBuilder()
.setUserName("Server Bot")
.setMessage("你好 " + request.getUserName() + "，我已收到你的消息：" + request.getMessage())
.setTimestamp(System.currentTimeMillis())
.build();
responseObserver.onNext(reply);
}
@Override
public void onError(Throwable t) {
logger.severe("Chat error: " + t.getMessage());
}
@Override
public void onCompleted() {
logger.info("Chat session ended");
responseObserver.onCompleted();
}
};
}
}
GrpcServer.java
package com.example.grpc.server;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
import java.util.logging.Logger;
public class GrpcServer {
private static final Logger logger = Logger.getLogger(GrpcServer.class.getName());
private static final int PORT = 50051;
private Server server;
private void start() throws IOException {
server = ServerBuilder.forPort(PORT)
.addService(new UserServiceImpl())
.build()
.start();
logger.info("gRPC Server started on port " + PORT);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
logger.info("Shutting down gRPC server...");
GrpcServer.this.stop();
logger.info("Server shut down");
}));
}
private void stop() {
if (server != null) {
server.shutdown();
}
}
private void blockUntilShutdown() throws InterruptedException {
if (server != null) {
server.awaitTermination();
}
}
public static void main(String[] args) throws IOException, InterruptedException {
GrpcServer grpcServer = new GrpcServer();
grpcServer.start();
grpcServer.blockUntilShutdown();
}
}
实现客户端
GrpcClient.java
package com.example.grpc.client;
import com.example.grpc.*;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
public class GrpcClient {
private static final Logger logger = Logger.getLogger(GrpcClient.class.getName());
private static final String HOST = "localhost";
private static final int PORT = 50051;
private final ManagedChannel channel;
private final UserServiceGrpc.UserServiceBlockingStub blockingStub;
private final UserServiceGrpc.UserServiceStub asyncStub;
public GrpcClient(String host, int port) {
this.channel = ManagedChannelBuilder.forAddress(host, port)
.usePlaintext()
.build();
this.blockingStub = UserServiceGrpc.newBlockingStub(channel);
this.asyncStub = UserServiceGrpc.newStub(channel);
}
public void shutdown() throws InterruptedException {
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
}
// 简单RPC示例
public void getUser(int userId) {
logger.info("Calling GetUser with userId: " + userId);
UserRequest request = UserRequest.newBuilder()
.setUserId(userId)
.build();
UserResponse response = blockingStub.getUser(request);
logger.info("Response - User: " + response.getName() +
", Age: " + response.getAge() +
", Email: " + response.getEmail());
}
// 服务器端流式RPC示例
public void listUsers(int page, int pageSize) {
logger.info("Calling ListUsers with page: " + page + ", size: " + pageSize);
UserListRequest request = UserListRequest.newBuilder()
.setPage(page)
.setPageSize(pageSize)
.build();
Iterator<UserResponse> responses = blockingStub.listUsers(request);
while (responses.hasNext()) {
UserResponse user = responses.next();
logger.info("Received user: " + user.getName() + " (ID: " + user.getUserId() + ")");
}
}
// 客户端流式RPC示例
public void createUsers() throws InterruptedException {
logger.info("Calling CreateUsers");
final CountDownLatch finishLatch = new CountDownLatch(1);
StreamObserver<CreateUserRequest> requestObserver = asyncStub.createUsers(
new StreamObserver<CreateUsersResponse>() {
@Override
public void onNext(CreateUsersResponse response) {
logger.info("Response: " + response.getMessage());
}
@Override
public void onError(Throwable t) {
logger.severe("Error: " + t.getMessage());
finishLatch.countDown();
}
@Override
public void onCompleted() {
logger.info("CreateUsers completed");
finishLatch.countDown();
}
}
);
// 发送多个用户创建请求
String[] names = {"李四", "王五", "赵六", "钱七"};
for (String name : names) {
CreateUserRequest request = CreateUserRequest.newBuilder()
.setName(name)
.setAge(20 + (int)(Math.random() * 30))
.setEmail(name + "@example.com")
.build();
requestObserver.onNext(request);
}
requestObserver.onCompleted();
finishLatch.await(1, TimeUnit.MINUTES);
}
// 双向流式RPC示例
public void chat() throws InterruptedException {
logger.info("Starting chat session");
final CountDownLatch finishLatch = new CountDownLatch(1);
StreamObserver<ChatMessage> requestObserver = asyncStub.chat(
new StreamObserver<ChatMessage>() {
@Override
public void onNext(ChatMessage response) {
logger.info("Server: " + response.getMessage());
}
@Override
public void onError(Throwable t) {
logger.severe("Chat error: " + t.getMessage());
finishLatch.countDown();
}
@Override
public void onCompleted() {
logger.info("Chat completed");
finishLatch.countDown();
}
}
);
// 发送聊天消息
String[] messages = {"你好", "今天天气不错", "再见"};
for (String message : messages) {
ChatMessage chatMessage = ChatMessage.newBuilder()
.setUserName("小明")
.setMessage(message)
.setTimestamp(System.currentTimeMillis())
.build();
requestObserver.onNext(chatMessage);
Thread.sleep(500);
}
requestObserver.onCompleted();
finishLatch.await(1, TimeUnit.MINUTES);
}
public static void main(String[] args) throws InterruptedException {
GrpcClient client = new GrpcClient(HOST, PORT);
try {
// 1. 简单RPC
logger.info("\n=== 简单RPC示例 ===");
client.getUser(1);
// 2. 服务器端流式RPC
logger.info("\n=== 服务器端流式RPC示例 ===");
client.listUsers(1, 3);
// 3. 客户端流式RPC
logger.info("\n=== 客户端流式RPC示例 ===");
client.createUsers();
// 4. 双向流式RPC
logger.info("\n=== 双向流式RPC示例 ===");
client.chat();
} finally {
client.shutdown();
}
}
}
运行测试
-
首先启动服务器：
mvn exec:java -Dexec.mainClass="com.example.grpc.server.GrpcServer"
-
然后运行客户端：
mvn exec:java -Dexec.mainClass="com.example.grpc.client.GrpcClient"
常见问题与最佳实践
错误处理示例
// 服务器端错误处理
@Override
public void getUser(UserRequest request, StreamObserver<UserResponse> responseObserver) {
try {
if (request.getUserId() <= 0) {
responseObserver.onError(
io.grpc.Status.INVALID_ARGUMENT
.withDescription("用户ID必须大于0")
.asRuntimeException()
);
return;
}
// 处理正常逻辑
UserResponse response = // ...
responseObserver.onNext(response);
responseObserver.onCompleted();
} catch (Exception e) {
responseObserver.onError(
io.grpc.Status.INTERNAL
.withDescription("服务器内部错误: " + e.getMessage())
.asRuntimeException()
);
}
}
SSL/TLS配置
// 服务器端启用SSL
Server server = ServerBuilder.forPort(PORT)
.useTransportSecurity(
new File("server.crt"),
new File("server.pem")
)
.addService(new UserServiceImpl())
.build();
// 客户端SSL连接
ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port)
.useTransportSecurity()
.build();
这个完整的案例覆盖了gRPC的四种通信模式：
- 简单RPC（Unary）
- 服务器端流式（Server Streaming）
- 客户端流式（Client Streaming）
- 双向流式（Bidirectional Streaming）
你可以根据实际需求选择相应的模式来实现你的功能。
