java版gRPC实战之四：客户端流

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容：所有原创文章分类汇总及配套源码，涉及Java、Docker、Kubernetes、DevOPS等；

《java版gRPC实战》全系列链接

  1. 用proto生成代码
  2. 服务发布和调用
  3. 服务端流
  4. 客户端流
  5. 双向流
  6. 客户端动态获取服务端地址
  7. 基于eureka的注册发现

本篇概览

  • 本文是《java版gRPC实战》系列的第四篇，前文掌握了服务端流，适合从服务端获取大量数据的场景，今天的目标是掌握客户端流类型的服务，包括服务提供方和使用方两侧的开发；
  • 先来看看官方资料对客户端流式RPC的介绍：客户端写入一个消息序列并将其发送到服务器，同样也是使用流。一旦客户端完成写入消息，它等待服务器完成读取返回它的响应；
  • 本文由以下几部分组成：
  1. 提前小结几个重要的知识点，稍后开发过程中要重点关注这几个地方；
  2. 在proto文件中定义客户端流类型的gRPC接口，再通过proto生成java代码；
  3. 开发服务端应用；
  4. 开发客户端应用；
  5. 验证；

提前小结

为了突出重点，这里将几个关键的知识点提前给出：

  1. 客户端流的特点，是请求方以流的形式提交数据到响应方；
  2. 一次RPC请求中，请求方可以通过流的方式源源不断的提交数据，直到调用了StreamObserver的onCompleted方法，才算提交数据完成；
  3. 平时咱们调用方法时，方法内部用到的数据是通过入参传进来的，但这里不一样，客户端要传给服务端的数据和gRPC方法的入参没有关系，而是和方法的返回对象有关（执行返回对象的onNext方法可以将数据传给服务端）；
  4. 客户端在A线程上传完数据后，服务端的响应是在另一个线程B执行的，因此，如果A线程拿到服务端响应，就要B线程的异步响应方法执行完毕，等待的方法有多种，我用的是CountDownLatch；
  5. 在服务端，开发者要编写的代码和以往web开发不同，不是将数据处理好返回，而是返回一个StreamObserver实例给上层框架，由框架负责处理的逻辑，开发者专注开发StreamObserver的实现即可，例如重写onNext方法，客户端通过流每上传一笔数据，onNext方法都会被外层框架执行一次；
  6. 如果您用的是IDEA，记得勾选下图红框中的选框，否则运行应用的时候可能遇到lombok相关的问题：
在这里插入图片描述
  • 上面提到的这些，会在接下来的开发过程中充分体现出来；

源码下载

名称 链接 备注
项目主页 https://github.com/zq2599/blog_demos 该项目在GitHub上的主页
git仓库地址(https) https://github.com/zq2599/blog_demos.git 该项目源码的仓库地址，https协议
git仓库地址(ssh) git@github.com:zq2599/blog_demos.git 该项目源码的仓库地址，ssh协议
  • 这个git项目中有多个文件夹，《java版gRPC实战》系列的源码在<font color="blue">grpc-tutorials</font>文件夹下，如下图红框所示：
在这里插入图片描述
  • <font color="blue">grpc-tutorials</font>文件夹下有多个目录，本篇文章对应的服务端代码在<font color="blue">client-stream-server-side</font>目录下，客户端代码在<font color="blue">client-stream-client-side</font>目录下，如下图：
在这里插入图片描述

在proto文件中定义客户端流类型的gRPC接口

  • 首先要做的就是定义gRPC接口，打开<font color="blue">mall.proto</font>，在里面新增方法和相关的数据结构，需要重点关注的是AddToCart方法的入参ProductOrder前面添加了<font color="red">stream</font>修饰，代表该方法是客户端流类型：
// gRPC服务，这是个在线商城的购物车服务
service CartService {
    // 客户端流式：添加多个商品到购物车
    rpc AddToCart (stream ProductOrder) returns (AddCartReply) {}
}

// 提交购物车时的产品信息
message ProductOrder {
    // 商品ID
    int32 productId = 1;
    // 商品数量
    int32 number = 2;
}

// 提交购物车返回结果的数据结构
message AddCartReply {
    // 返回码
    int32 code = 1;
    // 描述信息
    string message = 2;
}
  • 双击下图红框中的task即可生成java代码：
在这里插入图片描述
  • 生成下图红框中的文件：
在这里插入图片描述
  • 接下来开发服务端；

开发服务端应用

  • 在父工程grpc-turtorials下面新建名为<font color="blue">client-stream-server-side</font>的模块，其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')
    // annotationProcessor不会传递，使用了lombok生成代码的模块，需要自己声明annotationProcessor
    annotationProcessor 'org.projectlombok:lombok'
}
  • 配置文件application.yml：
spring:
  application:
    name: client-stream-server-side
# gRPC有关的配置，这里只需要配置服务端口号
grpc:
  server:
    port: 9900

  • 启动类ClientStreamServerSideApplication.java的代码就不贴了，普通的springboot启动类而已；

  • 重点是提供grpc服务的GrpcServerService.java，请结合前面小结的第五点来阅读代码，咱们要做的就是给上层框架返回一个匿名类，至于里面的onNext、onCompleted方法何时被调用是上层框架决定的，另外还准备了成员变量totalCount，这样就可以记录总数了：

package com.bolingcavalry.grpctutorials;

import com.bolingcavalry.grpctutorials.lib.AddCartReply;
import com.bolingcavalry.grpctutorials.lib.CartServiceGrpc;
import com.bolingcavalry.grpctutorials.lib.ProductOrder;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;

@GrpcService
@Slf4j
public class GrpcServerService extends CartServiceGrpc.CartServiceImplBase {

    @Override
    public StreamObserver<ProductOrder> addToCart(StreamObserver<AddCartReply> responseObserver) {
        // 返回匿名类，给上层框架使用
        return new StreamObserver<ProductOrder>() {

            // 记录处理产品的总量
            private int totalCount = 0;

            @Override
            public void onNext(ProductOrder value) {
                log.info("正在处理商品[{}]，数量为[{}]",
                        value.getProductId(),
                        value.getNumber());

                // 增加总量
                totalCount += value.getNumber();
            }

            @Override
            public void onError(Throwable t) {
                log.error("添加购物车异常", t);
            }

            @Override
            public void onCompleted() {
                log.info("添加购物车完成，共计[{}]件商品", totalCount);
                responseObserver.onNext(AddCartReply.newBuilder()
                                                    .setCode(10000)
                                                    .setMessage(String.format("添加购物车完成，共计[%d]件商品", totalCount))
                                                    .build());
                responseObserver.onCompleted();
            }
        };
    }
}

开发客户端应用

  • 在父工程grpc-turtorials下面新建名为<font color="blue">client-stream-server-side</font>的模块，其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，设置自己的web端口号和服务端地址：
server:
  port: 8082
spring:
  application:
    name: client-stream-client-side

grpc:
  client:
    # gRPC配置的名字，GrpcClient注解会用到
    client-stream-server-side:
      # gRPC服务端地址
      address: 'static://127.0.0.1:9900'
      enableKeepAlive: true
      keepAliveWithoutCalls: true
      negotiationType: plaintext
  • 启动类ClientStreamClientSideApplication.java的代码就不贴了，普通的springboot启动类而已；
  • 正常情况下我们都是用StreamObserver处理服务端响应，这里由于是异步响应，需要额外的方法从StreamObserver中取出业务数据，于是定一个新接口，继承自StreamObserver，新增<font color="blue">getExtra</font>方法可以返回String对象，详细的用法稍后会看到：
package com.bolingcavalry.grpctutorials;

import io.grpc.stub.StreamObserver;

public interface ExtendResponseObserver<T> extends StreamObserver<T> {
    String getExtra();
}
  • 重头戏来了，看看如何远程调用客户端流类型的gRPC接口，前面小结提到的2、3、4点都会涉及到，代码中已经添加详细注释：
package com.bolingcavalry.grpctutorials;

import com.bolingcavalry.grpctutorials.lib.AddCartReply;
import com.bolingcavalry.grpctutorials.lib.CartServiceGrpc;
import com.bolingcavalry.grpctutorials.lib.ProductOrder;
import io.grpc.stub.StreamObserver;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.stereotype.Service;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class GrpcClientService {

    @GrpcClient("client-stream-server-side")
    private CartServiceGrpc.CartServiceStub cartServiceStub;

    public String addToCart(int count) {
        
        CountDownLatch countDownLatch = new CountDownLatch(1);
        
        // responseObserver的onNext和onCompleted会在另一个线程中被执行，
        // ExtendResponseObserver继承自StreamObserver
        ExtendResponseObserver<AddCartReply> responseObserver = new ExtendResponseObserver<AddCartReply>() {

            String extraStr;

            @Override
            public String getExtra() {
                return extraStr;
            }

            private int code;

            private String message;

            @Override
            public void onNext(AddCartReply value) {
                log.info("on next");
                code = value.getCode();
                message = value.getMessage();
            }

            @Override
            public void onError(Throwable t) {
                log.error("gRPC request error", t);
                extraStr = "gRPC error, " + t.getMessage();
                countDownLatch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("on complete");
                extraStr = String.format("返回码[%d]，返回信息:%s" , code, message);
                countDownLatch.countDown();
            }
        };
        
        // 远程调用，此时数据还没有给到服务端
        StreamObserver<ProductOrder> requestObserver = cartServiceStub.addToCart(responseObserver);
        
        for(int i=0; i<count; i++) {
            // 发送一笔数据到服务端
            requestObserver.onNext(build(101 + i, 1 + i));
        }

        // 客户端告诉服务端：数据已经发完了
        requestObserver.onCompleted();

        try {
            // 开始等待，如果服务端处理完成，那么responseObserver的onCompleted方法会在另一个线程被执行，
            // 那里会执行countDownLatch的countDown方法，一但countDown被执行，下面的await就执行完毕了，
            // await的超时时间设置为2秒
            countDownLatch.await(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("countDownLatch await error", e);
        }

        log.info("service finish");
        // 服务端返回的内容被放置在requestObserver中，从getExtra方法可以取得
        return responseObserver.getExtra();
    }

    /**
     * 创建ProductOrder对象
     * @param productId
     * @param num
     * @return
     */
    private static ProductOrder build(int productId, int num) {
        return ProductOrder.newBuilder().setProductId(productId).setNumber(num).build();
    }
}
  • 最后做个web接口，可以通过web请求验证远程调用：
package com.bolingcavalry.grpctutorials;

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;

@RestController
public class GrpcClientController {

    @Autowired
    private GrpcClientService grpcClientService;

    @RequestMapping("/")
    public String printMessage(@RequestParam(defaultValue = "1") int count) {
        return grpcClientService.addToCart(count);
    }
}
  • 编码完成，开始验证；

验证

  • 启动服务端ClientStreamServerSideApplication：
在这里插入图片描述
  • 启动客户端ClientStreamClientSideApplication：
在这里插入图片描述
在这里插入图片描述
  • 下面是服务端日志，可见逐一处理了客户端的每一笔数据：
在这里插入图片描述
  • 下面是客户端日志，可见由于CountDownLatch的作用，发起gRPC请求的线程一直等待responseObserver.onCompleted在另一个线程被执行完后，才会继续执行：
在这里插入图片描述
  • 至此，客户端流类型的gRPC服务及其客户端开发就完成了，这种异步操作与咱们平时开发同步类型的web接口还是有差别的，希望本文能给您带来一些参考，下一篇咱们实战最后一种类型：双向流式；

你不孤单，欣宸原创一路相伴

  1. Java系列
  2. Spring系列
  3. Docker系列
  4. kubernetes系列
  5. 数据库+中间件系列
  6. DevOps系列

欢迎关注公众号：程序员欣宸

微信搜索「程序员欣宸」，我是欣宸，期待与您一同畅游Java世界...
https://github.com/zq2599/blog_demos

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 217,826评论 6赞 506
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 92,968评论 3赞 395
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 164,234评论 0赞 354
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 58,562评论 1赞 293
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 67,611评论 6赞 392
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,482评论 1赞 302
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,271评论 3赞 418
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,166评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,608评论 1赞 314
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,814评论 3赞 336
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,926评论 1赞 348
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,644评论 5赞 346
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,249评论 3赞 329
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,866评论 0赞 22
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,991评论 1赞 269
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,063评论 3赞 370
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,871评论 2赞 354

推荐阅读更多精彩内容