java 版 gRPC 实战之三：服务端流

作者：程序员欣宸

2022 年 3 月 30 日
本文字数：5051 字
阅读完需：约 17 分钟

欢迎访问我的 GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos

关于 gRPC 定义的四种类型

本文是《java 版 gRPC 实战》系列的第三篇，前文咱们实战体验了简单的 RPC 请求和响应，那种简单的请求响应方式其实只是 gRPC 定义的四种类型之一，这里给出《gRPC 官方文档中文版》对这四种 gRPC 类型的描述：

简单 RPC：客户端使用存根(stub)发送请求到服务器并等待响应返回，就像平常的函数调用一样；
服务器端流式 RPC：客户端发送请求到服务器，拿到一个流去读取返回的消息序列。客户端读取返回的流，直到里面没有任何消息；(即本篇内容)
客户端流式 RPC：客户端写入一个消息序列并将其发送到服务器，同样也是使用流。一旦客户端完成写入消息，它等待服务器完成读取返回它的响应；
双向流式 RPC：是双方使用读写流去发送一个消息序列。两个流独立操作，因此客户端和服务器可以以任意喜欢的顺序读写：比如，服务器可以在写入响应前等待接收所有的客户端消息，或者可以交替的读取和写入消息，或者其他读写的组合。每个流中的消息顺序被预留；

本篇概览

本篇是服务端流类型的 gRPC 服务实战，包括以下内容：

开发一个 gRPC 服务，类型是服务端流；
开发一个客户端，调用前面发布的 gRPC 服务；
验证；

不多说了，开始上代码；

源码下载

本篇实战中的完整源码可在 GitHub 下载到，地址和链接信息如下表所示(https://github.com/zq2599/blog_demos)：

这个 git 项目中有多个文件夹，《java 版 gRPC 实战》系列的源码在 grpc-tutorials 文件夹下，如下图红框所示：
grpc-tutorials 文件夹下有多个目录，本篇文章对应的服务端代码在 server-stream-server-side 目录下，客户端代码在 server-stream-client-side 目录下，如下图：

开发一个 gRPC 服务，类型是服务端流

首先要开发的是 gRPC 服务端，一共要做下图所示的七件事：
打开 grpc-lib 模块，在 src/main/proto 目录下新增文件 mall.proto，里面定一个了一个 gRPC 方法 ListOrders 及其入参和返回对象，内容如下，要注意的是返回值要用关键字 stream 修饰，表示该接口类型是服务端流：

syntax = "proto3";
option java_multiple_files = true;// 生成java代码的packageoption java_package = "com.bolingcavalry.grpctutorials.lib";// 类名option java_outer_classname = "MallProto";
// gRPC服务，这是个在线商城的订单查询服务service OrderQuery {    // 服务端流式：订单列表接口，入参是买家信息，返回订单列表(用stream修饰返回值)    rpc ListOrders (Buyer) returns (stream Order) {}}
// 买家IDmessage Buyer {    int32 buyerId = 1;}
// 返回结果的数据结构message Order {    // 订单ID    int32 orderId = 1;    // 商品ID    int32 productId = 2;    // 交易时间    int64 orderTime = 3;    // 买家备注    string buyerRemark = 4;}

复制代码

双击下图红框位置的 generateProto，即可根据 proto 生成 java 代码：
新生成的 java 代码如下图红框：
在父工程 grpc-turtorials 下面新建名为 server-stream-server-side 的模块，其 build.gradle 内容如下：

// 使用springboot插件plugins {    id 'org.springframework.boot'}
dependencies {    implementation 'org.projectlombok:lombok'    implementation 'org.springframework.boot:spring-boot-starter'    // 作为gRPC服务提供方，需要用到此库    implementation 'net.devh:grpc-server-spring-boot-starter'    // 依赖自动生成源码的工程    implementation project(':grpc-lib')}

复制代码

新建配置文件 application.yml：

spring:  application:    name: server-stream-server-side# gRPC有关的配置，这里只需要配置服务端口号grpc:  server:    port: 9899

复制代码

启动类：

package com.bolingcavalry.grpctutorials;
import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplicationpublic class ServerStreamServerSideApplication {    public static void main(String[] args) {        SpringApplication.run(ServerStreamServerSideApplication.class, args);    }}

复制代码

接下来是最关键的 gRPC 服务，代码如下，可见 responseObserver.onNext 方法被多次调用，用以向客户端持续输出数据，最后通过 responseObserver.onCompleted 结束输出：

package com.bolingcavalry.grpctutorials;
import com.bolingcavalry.grpctutorials.lib.Buyer;import com.bolingcavalry.grpctutorials.lib.Order;import com.bolingcavalry.grpctutorials.lib.OrderQueryGrpc;import io.grpc.stub.StreamObserver;import net.devh.boot.grpc.server.service.GrpcService;import java.util.ArrayList;import java.util.List;
@GrpcServicepublic class GrpcServerService extends OrderQueryGrpc.OrderQueryImplBase {
    /**     * mock一批数据     * @return     */    private static List<Order> mockOrders(){        List<Order> list = new ArrayList<>();        Order.Builder builder = Order.newBuilder();
        for (int i = 0; i < 10; i++) {            list.add(builder                    .setOrderId(i)                    .setProductId(1000+i)                    .setOrderTime(System.currentTimeMillis()/1000)                    .setBuyerRemark(("remark-" + i))                    .build());        }
        return list;    }
    @Override    public void listOrders(Buyer request, StreamObserver<Order> responseObserver) {        // 持续输出到client        for (Order order : mockOrders()) {            responseObserver.onNext(order);        }        // 结束输出        responseObserver.onCompleted();    }}

复制代码

至此，服务端开发完成，咱们再开发一个 springboot 应用作为客户端，看看如何远程调用 listOrders 接口，得到 responseObserver.onNext 方法输出的数据；

开发一个客户端，调用前面发布的 gRPC 服务

客户端模块的基本功能是提供一个 web 接口，其内部会调用服务端的 listOrders 接口，将得到的数据返回给前端，如下图：
在父工程 grpc-turtorials 下面新建名为 server-stream-client-side 的模块，其 build.gradle 内容如下：

plugins {    id 'org.springframework.boot'}
dependencies {    implementation 'org.projectlombok:lombok'    implementation 'org.springframework.boot:spring-boot-starter'    implementation 'org.springframework.boot:spring-boot-starter-web'    implementation 'net.devh:grpc-client-spring-boot-starter'    implementation project(':grpc-lib')}

复制代码

应用配置信息 application.yml 内容如下，可见是端口和 gRPC 服务端地址的配置：

server:  port: 8081spring:  application:    name: server-stream-client-side
grpc:  client:    # gRPC配置的名字，GrpcClient注解会用到    server-stream-server-side:      # gRPC服务端地址      address: 'static://127.0.0.1:9899'      enableKeepAlive: true      keepAliveWithoutCalls: true      negotiationType: plaintext

复制代码

服务端的 listOrders 接口返回的 Order 对象里面有很多 gRPC 相关的内容，不适合作为 web 接口的返回值，因此定义一个 DispOrder 类作为 web 接口返回值：

package com.bolingcavalry.grpctutorials;
import lombok.AllArgsConstructor;import lombok.Data;import java.io.Serializable;
@Data@AllArgsConstructorpublic class DispOrder {    private int orderId;    private int productId;    private String orderTime;    private String buyerRemark;}

复制代码

平淡无奇的启动类：

package com.bolingcavalry.grpctutorials;
import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplicationpublic class ServerStreamClientSideApplication {
    public static void main(String[] args) {        SpringApplication.run(ServerStreamClientSideApplication.class, args);    }}

复制代码

重点来了，GrpcClientService.java，里面展示了如何远程调用 gRPC 服务的 listOrders 接口，可见对于服务端流类型的接口，客户端这边通过 stub 调用会得到 Iterator 类型的返回值，接下来要做的就是遍历 Iterator：

package com.bolingcavalry.grpctutorials;
import com.bolingcavalry.grpctutorials.lib.Buyer;import com.bolingcavalry.grpctutorials.lib.Order;import com.bolingcavalry.grpctutorials.lib.OrderQueryGrpc;import io.grpc.StatusRuntimeException;import lombok.extern.slf4j.Slf4j;import net.devh.boot.grpc.client.inject.GrpcClient;import org.springframework.stereotype.Service;import java.time.Instant;import java.time.LocalDateTime;import java.time.ZoneOffset;import java.time.format.DateTimeFormatter;import java.util.ArrayList;import java.util.Iterator;import java.util.List;
@Service@Slf4jpublic class GrpcClientService {
    @GrpcClient("server-stream-server-side")    private OrderQueryGrpc.OrderQueryBlockingStub orderQueryBlockingStub;
    public List<DispOrder> listOrders(final String name) {        // gRPC的请求参数        Buyer buyer = Buyer.newBuilder().setBuyerId(101).build();
        // gRPC的响应        Iterator<Order> orderIterator;
        // 当前方法的返回值        List<DispOrder> orders = new ArrayList<>();
        // 通过stub发起远程gRPC请求        try {            orderIterator = orderQueryBlockingStub.listOrders(buyer);        } catch (final StatusRuntimeException e) {            log.error("error grpc invoke", e);            return new ArrayList<>();        }
        DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        log.info("start put order to list");        while (orderIterator.hasNext()) {            Order order = orderIterator.next();
            orders.add(new DispOrder(order.getOrderId(),                                    order.getProductId(),                                    // 使用DateTimeFormatter将时间戳转为字符串                                    dtf.format(LocalDateTime.ofEpochSecond(order.getOrderTime(), 0, ZoneOffset.of("+8"))),                                    order.getBuyerRemark()));            log.info("");        }
        log.info("end put order to list");
        return orders;    }}

复制代码

最后做一个 controller 类，对外提供一个 web 接口，里面会调用 GrpcClientService 的方法：

package com.bolingcavalry.grpctutorials;
import com.bolingcavalry.grpctutorials.lib.Order;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RestController;import java.util.List;
@RestControllerpublic class GrpcClientController {
    @Autowired    private GrpcClientService grpcClientService;
    @RequestMapping("/")    public List<DispOrder> printMessage(@RequestParam(defaultValue = "will") String name) {        return grpcClientService.listOrders(name);    }}

复制代码

至此，编码完成，开始验证

验证

启动 server-stream-server-side，启动成功后会监听 9989 端口：
启动 server-stream-client-side，再在浏览器访问：http://localhost:8081/?name=Tom ，得到结果如下(firefox 自动格式化 json 数据)，可见成功地获取了 gRPC 的远程数据：
至此，服务端流类型的 gRPC 接口的开发和使用实战就完成了，接下来的章节还会继续学习另外两种类型；