歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

內容：所有原創文章分類彙總及配套原始碼，涉及Java、Docker、Kubernetes、DevOPS等；

本篇概覽

  • 本文是《java版gRPC實戰》系列的第五篇，目標是掌握雙向流型別的服務，即請求引數是流的形式，響應的內容也是流的形式；
  • 先來看看官方資料對雙向流式RPC的介紹：是雙方使用讀寫流去傳送一個訊息序列。兩個流獨立操作，因此客戶端和伺服器 可以以任意喜歡的順序讀寫：比如， 伺服器可以在寫入響應前等待接收所有的客戶端訊息，或者可以交替 的讀取和寫入訊息，或者其他讀寫的組合。 每個流中的訊息順序被預留；
  • 掌握了客戶端流和服務端流兩種型別的開發後，雙向流型別就很好理解了，就是之前兩種型別的結合體，請求和響應都按照流的方式處理即可；
  • 今天的實戰，咱們來設計一個線上商城的功能：批量減扣庫存，即客戶端提交多個商品和數量，服務端返回每個商品減扣庫存成功和失敗的情況；
  • 咱們儘快進入編碼環節吧，具體內容如下：
  1. 在proto檔案中定義雙向流型別的gRPC介面，再通過proto生成java程式碼
  2. 開發服務端應用
  3. 開發客戶端應用
  4. 驗證

原始碼下載

名稱 連結 備註
專案主頁 https://github.com/zq2599/blog_demos 該專案在GitHub上的主頁
git倉庫地址(https) https://github.com/zq2599/blog_demos.git 該專案原始碼的倉庫地址，https協議
git倉庫地址(ssh) [email protected]:zq2599/blog_demos.git 該專案原始碼的倉庫地址，ssh協議
  • 這個git專案中有多個資料夾，《java版gRPC實戰》系列的原始碼在grpc-tutorials資料夾下，如下圖紅框所示：

  • grpc-tutorials資料夾下有多個目錄，本篇文章對應的服務端程式碼在double-stream-server-side目錄下，客戶端程式碼在double-stream-client-side目錄下，如下圖：

在proto檔案中定義雙向流型別的gRPC介面

  • 首先要做的就是定義gRPC介面，開啟mall.proto，在裡面新增方法和相關的資料結構，需要重點關注的是BatchDeduct方法的入參ProductOrder和返回值DeductReply都添加了stream修飾（ProductOrder是上一章定義的），代表該方法是雙向流型別：
// gRPC服務，這是個線上商城的庫存服務

service StockService {

    // 雙向流式：批量扣減庫存

    rpc BatchDeduct (stream ProductOrder) returns (stream DeductReply) {}

}

// 扣減庫存返回結果的資料結構

message DeductReply {

    // 返回碼

    int32 code = 1;

    // 描述資訊

    string message = 2;

}
  • 雙擊下圖紅框中的task即可生成java程式碼：

  • 生成下圖紅框中的檔案，即服務端定義和返回值資料結構：

  • 接下來開發服務端；

開發服務端應用

  • 在父工程grpc-turtorials下面新建名為double-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')

    // annotationProcessor不會傳遞，使用了lombok生成程式碼的模組，需要自己宣告annotationProcessor

    annotationProcessor 'org.projectlombok:lombok'

}
  • 配置檔案application.yml：
spring:

  application:

    name: double-stream-server-side

# gRPC有關的配置，這裡只需要配置服務埠號

grpc:

  server:

    port: 9901
  • 啟動類DoubleStreamServerSideApplication.java的程式碼就不貼了，普通的springboot啟動類而已；
  • 重點是提供grpc服務的GrpcServerService.java，咱們要做的就是給上層框架返回一個匿名類，至於裡面的onNext、onCompleted方法何時被呼叫是上層框架決定的，另外還準備了成員變數totalCount，這樣就可以記錄總數了，由於請求引數是流，因此匿名類的onNext會被多次呼叫，並且由於返回值是流，因此onNext中呼叫了responseObserver.onNext方法來響應流中的每個請求，這樣客戶端就不斷收到服務端的響應資料(即客戶端的onNext方法會被多次呼叫)：
package grpctutorials;

import com.bolingcavalry.grpctutorials.lib.DeductReply;

import com.bolingcavalry.grpctutorials.lib.ProductOrder;

import com.bolingcavalry.grpctutorials.lib.StockServiceGrpc;

import io.grpc.stub.StreamObserver;

import lombok.extern.slf4j.Slf4j;

import net.devh.boot.grpc.server.service.GrpcService;

@GrpcService

@Slf4j

public class GrpcServerService extends StockServiceGrpc.StockServiceImplBase {

    @Override

    public StreamObserver<ProductOrder> batchDeduct(StreamObserver<DeductReply> responseObserver) {

        // 返回匿名類，給上層框架使用

        return new StreamObserver<ProductOrder>() {

            private int totalCount = 0;

            @Override

            public void onNext(ProductOrder value) {

                log.info("正在處理商品[{}]，數量為[{}]",

                        value.getProductId(),

                        value.getNumber());

                // 增加總量

                totalCount += value.getNumber();

                int code;

                String message;

                // 假設單數的都有庫存不足的問題

                if (0 == value.getNumber() % 2) {

                    code = 10000;

                    message = String.format("商品[%d]扣減庫存數[%d]成功", value.getProductId(), value.getNumber());

                } else {

                    code = 10001;

                    message = String.format("商品[%d]扣減庫存數[%d]失敗", value.getProductId(), value.getNumber());

                }

                responseObserver.onNext(DeductReply.newBuilder()

                        .setCode(code)

                        .setMessage(message)

                        .build());

            }

            @Override

            public void onError(Throwable t) {

                log.error("批量減扣庫存異常", t);

            }

            @Override

            public void onCompleted() {

                log.info("批量減扣庫存完成，共計[{}]件商品", totalCount);

                responseObserver.onCompleted();

            }

        };

    }

}

開發客戶端應用

  • 在父工程grpc-turtorials下面新建名為double-stream-server-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，設定自己的web埠號和服務端地址：
server:

  port: 8082

spring:

  application:

    name: double-stream-client-side

grpc:

  client:

    # gRPC配置的名字，GrpcClient註解會用到

    double-stream-server-side:

      # gRPC服務端地址

      address: 'static://127.0.0.1:9901'

      enableKeepAlive: true

      keepAliveWithoutCalls: true

      negotiationType: plaintext

  • 啟動類DoubleStreamClientSideApplication.java的程式碼就不貼了，普通的springboot啟動類而已；

  • 正常情況下我們都是用StreamObserver處理服務端響應，這裡由於是非同步響應，需要額外的方法從StreamObserver中取出業務資料，於是定一個新介面，繼承自StreamObserver，新增getExtra方法可以返回String物件，詳細的用法稍後會看到：

package com.bolingcavalry.grpctutorials;

import io.grpc.stub.StreamObserver;

public interface ExtendResponseObserver<T> extends StreamObserver<T> {

    String getExtra();

}
  • 重頭戲來了，看看如何遠端呼叫雙向流型別的gRPC介面，程式碼中已經新增詳細註釋：
package grpctutorials;

import com.bolingcavalry.grpctutorials.lib.DeductReply;

import com.bolingcavalry.grpctutorials.lib.ProductOrder;

import com.bolingcavalry.grpctutorials.lib.StockServiceGrpc;

import io.grpc.stub.StreamObserver;

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("double-stream-server-side")

    private StockServiceGrpc.StockServiceStub stockServiceStub;

    /**

     * 批量減庫存

     * @param count

     * @return

     */

    public String batchDeduct(int count) {

        CountDownLatch countDownLatch = new CountDownLatch(1);

        // responseObserver的onNext和onCompleted會在另一個執行緒中被執行，

        // ExtendResponseObserver繼承自StreamObserver

        ExtendResponseObserver<DeductReply> responseObserver = new ExtendResponseObserver<DeductReply>() {

            // 用stringBuilder儲存所有來自服務端的響應

            private StringBuilder stringBuilder = new StringBuilder();

            @Override

            public String getExtra() {

                return stringBuilder.toString();

            }

            /**

             * 客戶端的流式請求期間，每一筆請求都會收到服務端的一個響應，

             * 對應每個響應，這裡的onNext方法都會被執行一次，入參是響應內容

             * @param value

             */

            @Override

            public void onNext(DeductReply value) {

                log.info("batch deduct on next");

                // 放入匿名類的成員變數中

                stringBuilder.append(String.format("返回碼[%d]，返回資訊:%s<br>" , value.getCode(), value.getMessage()));

            }

            @Override

            public void onError(Throwable t) {

                log.error("batch deduct gRPC request error", t);

                stringBuilder.append("batch deduct gRPC error, " + t.getMessage());

                countDownLatch.countDown();

            }

            /**

             * 服務端確認響應完成後，這裡的onCompleted方法會被呼叫

             */

            @Override

            public void onCompleted() {

                log.info("batch deduct on complete");

                // 執行了countDown方法後，前面執行countDownLatch.await方法的執行緒就不再wait了，

                // 會繼續往下執行

                countDownLatch.countDown();

            }

        };

        // 遠端呼叫，此時資料還沒有給到服務端

        StreamObserver<ProductOrder> requestObserver = stockServiceStub.batchDeduct(responseObserver);

        for(int i=0; i<count; i++) {

            // 每次執行onNext都會發送一筆資料到服務端，

            // 服務端的onNext方法都會被執行一次

            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 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;

@RestController

public class GrpcClientController {

    @Autowired

    private GrpcClientService grpcClientService;

    @RequestMapping("/")

    public String printMessage(@RequestParam(defaultValue = "1") int count) {

        return grpcClientService.batchDeduct(count);

    }

}
  • 編碼完成，開始驗證；

驗證

  • 啟動服務端DoubleStreamServerSideApplication：

  • 啟動客戶端DoubleStreamClientSideApplication：

  • 這裡要改：瀏覽器輸入http://localhost:8083/?count=10，響應如下，可見遠端呼叫gRPC服務成功，流式響應的每一筆返回都被客戶端收到：

  • 下面是服務端日誌，可見逐一處理了客戶端的每一筆資料：

  • 下面是客戶端日誌，可見由於CountDownLatch的作用，發起gRPC請求的執行緒一直等待responseObserver.onCompleted在另一個執行緒被執行完後，才會繼續執行：

  • 至此，四種類型的gRPC服務及其客戶端開發就完成了，一般的業務場景咱們都能應付自如，接下來的文章咱們會繼續深入學習，瞭解複雜場景下的gRPC操作；

你不孤單，欣宸原創一路相伴

  1. Java系列
  2. Spring系列
  3. Docker系列
  4. kubernetes系列
  5. 資料庫+中介軟體系列
  6. DevOps系列

歡迎關注公眾號：程式設計師欣宸

微信搜尋「程式設計師欣宸」，我是欣宸，期待與您一同暢遊Java世界...

https://github.com/zq2599/blog_demos