IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Java案例如何实现gRPC？

wen python案例 172

本文目录导读：

Java案例如何实现gRPC？

  1. 环境准备
  2. 定义Protobuf服务
  3. 编译生成代码
  4. 实现服务器端
  5. 实现客户端
  6. 运行测试
  7. 常见问题与最佳实践

我来详细介绍如何在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();
        }
    }
}

运行测试

  1. 首先启动服务器：

    mvn exec:java -Dexec.mainClass="com.example.grpc.server.GrpcServer"

  2. 然后运行客户端：

    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）

你可以根据实际需求选择相应的模式来实现你的功能。

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Java案例如何实现GraphQL？

下一篇Spring Boot怎么写REST？

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝