Java版gRPC的使用之二：服务端流、客户端流和双向流

前言

在前文的基础上对客户端流、服务端流和双向流进行简单的应用：
Java版gRPC的使用之一：简单的gRPC搭建

关于gRPC定义的四种类型

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

两种Stub

Stub：异步的stub，可以支持所有的服务调用；

BlockingStub：阻塞式的sutb，可以支持一元和流式输出的服务调用

完整proto文件

syntax = "proto3"; // 协议版本

// 选项配置
option java_multiple_files = true;
option java_package = "com.example.springbootgrpclib.grpc.protobuf";
option java_outer_classname = "SimpleProto";

service Simple {
    // 简单gRPC
    rpc OneToOne (MyRequest) returns (MyResponse) {
    }
    //服务端流
    rpc OneToMany (MyRequest) returns(stream MyResponse);

    //客户端流
    rpc ManyToOne (stream MyRequest) returns(MyResponse);

    //双向流
    rpc ManyToMany (stream MyRequest) returns(stream MyResponse);
}

message MyRequest {
    string name = 1;

    int32 value = 2;
}

message MyResponse {
    string message = 1;

    int64 result = 2;
}

一、服务端流 gRPC服务端代码

package com.example.localserver.server;

import com.example.springbootgrpclib.grpc.protobuf.MyRequest;
import com.example.springbootgrpclib.grpc.protobuf.MyResponse;
import com.example.springbootgrpclib.grpc.protobuf.SimpleGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;

import java.util.Date;

@GrpcService
@Slf4j
public class GrpcServerService extends SimpleGrpc.SimpleImplBase {

    //服务端流
    @Override
    public void oneToMany(MyRequest request, StreamObserver responseObserver) {
        log.info("接收客户端数据{}", request);
        long result = 0;
        for (int i = 1; i <= 10; i++) {
            MyResponse myResponse = MyResponse.newBuilder()
                    .setMessage(request.getName() + i)
                    .setResult(result += i)
                    .build();
            responseObserver.onNext(myResponse);
        }
        responseObserver.onCompleted();
    }
}

可见responseObserver.onNext方法被多次调用，用以向客户端持续输出数据，最后通过responseObserver.onCompleted结束输出：

gRPC客户端代码

package com.example.localclient.service;

import com.example.springbootgrpclib.grpc.protobuf.MyRequest;
import com.example.springbootgrpclib.grpc.protobuf.MyResponse;
import com.example.springbootgrpclib.grpc.protobuf.SimpleGrpc;
import io.grpc.StatusRuntimeException;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.stereotype.Service;


@Service
@Slf4j
public class GrpcClientService {

    @GrpcClient("local-grpc-server")
    private SimpleGrpc.SimpleBlockingStub simpleStub;

    public List oneToMany(String name) {
        List result = new ArrayList<>();
        Iterator myResponseIterator;
        try{
            myResponseIterator = this.simpleStub.oneToMany(MyRequest.newBuilder().setName(name).build());
        }catch (final StatusRuntimeException e) {
            return new ArrayList<>();
        }
        log.info("start put list");
        while (myResponseIterator.hasNext()){
            MyResponse myResponse = myResponseIterator.next();
            log.info("result = "+ myResponse.getResult());
            result.add(myResponse.getMessage());
        }
        log.info("end put list");
        return result;
    }
}

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

二、客户端流

客户端流的特点，是请求方以流的形式提交数据到响应方；
一次RPC请求中，请求方可以通过流的方式源源不断的提交数据，直到调用了StreamObserver的onCompleted方法，才算提交数据完成；
平时咱们调用方法时，方法内部用到的数据是通过入参传进来的，但这里不一样，客户端要传给服务端的数据和gRPC方法的入参没有关系，而是和方法的返回对象有关（执行返回对象的onNext方法可以将数据传给服务端）；
客户端在A线程上传完数据后，服务端的响应是在另一个线程B执行的，因此，如果A线程拿到服务端响应，就要B线程的异步响应方法执行完毕，等待的方法有多种，我用的是CountDownLatch；
在服务端，开发者要编写的代码和以往web开发不同，不是将数据处理好返回，而是返回一个StreamObserver实例给上层框架，由框架负责处理的逻辑，开发者专注开发StreamObserver的实现即可，例如重写onNext方法，客户端通过流每上传一笔数据，onNext方法都会被外层框架执行一次；

gRPC服务端代码

package com.example.localserver.server;

import com.example.springbootgrpclib.grpc.protobuf.MyRequest;
import com.example.springbootgrpclib.grpc.protobuf.MyResponse;
import com.example.springbootgrpclib.grpc.protobuf.SimpleGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;


@GrpcService
@Slf4j
public class GrpcServerService extends SimpleGrpc.SimpleImplBase {

    @Override
    public void oneToOne(MyRequest request, StreamObserver responseObserver) {
        log.info("接收客户端数据{}", request);
        MyResponse response = MyResponse.newBuilder().setMessage( request.getName()).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    //服务端流
    @Override
    public void oneToMany(MyRequest request, StreamObserver responseObserver) {
        log.info("接收客户端数据{}", request);
        long result = 0;
        for (int i = 1; i <= 10; i++) {
            MyResponse myResponse = MyResponse.newBuilder()
                    .setMessage(request.getName() + i)
                    .setResult(result += i)
                    .build();
            responseObserver.onNext(myResponse);
        }
        responseObserver.onCompleted();
    }

    //客户端流
    @Override
    public StreamObserver manyToOne(StreamObserver responseObserver) {
        // 返回匿名类，给上层框架使用
        return new StreamObserver() {

            // 记录总量
            private int totalCount = 0;

            @Override
            public void onNext(MyRequest myRequest) {
                log.info("正在处理[{}]，数量为[{}]",
                        myRequest.getName(),
                        myRequest.getValue());

                // 增加总量
                totalCount += myRequest.getValue();
            }

            @Override
            public void onError(Throwable t) {
                log.error("异常", t);
            }

            @Override
            public void onCompleted() {
                log.info("共计[{}]", totalCount);
                responseObserver.onNext(MyResponse.newBuilder()
                        .setResult(totalCount)
                        .setMessage(String.format("共计[%d]", totalCount))
                        .build());
                responseObserver.onCompleted();
            }
        };
    }

    //双向流
    @Override
    public StreamObserver manyToMany(StreamObserver responseObserver) {
        // 返回匿名类，给上层框架使用
        return new StreamObserver() {

            // 记录总量
            private int totalCount = 0;

            @Override
            public void onNext(MyRequest myRequest) {
                log.info("正在处理[{}]，数量为[{}]",
                        myRequest.getName(),
                        myRequest.getValue());

                // 增加总量
                totalCount += myRequest.getValue();

                Integer value;
                String message;
                if (0 == myRequest.getValue() % 2) {
                    value = myRequest.getValue() / 2;
                    message = String.format(myRequest.getName() + "的值是2的倍数，减半[%d]成功", value);
                } else {
                    value = (myRequest.getValue() + 1) / 2;
                    message = String.format(myRequest.getName() + "的值不是2的倍数，减半[%d]成功", value);
                }

                responseObserver.onNext(MyResponse.newBuilder()
                        .setResult(value)
                        .setMessage(message)
                        .build());
            }

            @Override
            public void onError(Throwable t) {
                log.error("异常", t);
            }

            @Override
            public void onCompleted() {
                log.info("共计[{}]", totalCount);
                responseObserver.onCompleted();
            }
        };
    }
}

gRPC客户端代码

正常情况下我们都是用StreamObserver处理服务端响应，这里由于是异步响应，需要额外的方法从StreamObserver中取出业务数据，于是定一个新接口，继承自StreamObserver，新增getExtra方法可以返回String对象

package com.example.localclient.service;

import io.grpc.stub.StreamObserver;

public interface ExtendResponseObserver extends StreamObserver {
    Object getExtra();

}

具体代码

package com.example.localclient.service;

import com.example.springbootgrpclib.grpc.protobuf.MyRequest;
import com.example.springbootgrpclib.grpc.protobuf.MyResponse;
import com.example.springbootgrpclib.grpc.protobuf.SimpleGrpc;
import io.grpc.StatusRuntimeException;
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.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class GrpcClientService {

    @GrpcClient("local-grpc-server")
    private SimpleGrpc.SimpleStub stub;

    public String manyToOne(String name,Integer value){
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // responseObserver的onNext和onCompleted会在另一个线程中被执行，
        // ExtendResponseObserver继承自StreamObserver
        ExtendResponseObserver responseObserver = new ExtendResponseObserver() {

            String extraStr;

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

            private long result;
            private String message;

            @Override
            public void onNext(MyResponse myResponse) {
                log.info("on next");
                result = myResponse.getResult();
                message = myResponse.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" , result, message);
                countDownLatch.countDown();
            }
        };

        // 远程调用，此时数据还没有给到服务端
        StreamObserver requestObserver = stub.manyToOne(responseObserver);

        for(int i=1; i <= 10; i++) {
            // 发送一笔数据到服务端
            requestObserver.onNext(MyRequest.newBuilder()
                    .setName(name)
                    .setValue(value + i)
                    .build());
        }

        // 客户端告诉服务端：数据已经发完了
        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 (String) responseObserver.getExtra();
    }

}

三、双向流 gRPC服务端代码

在GrpcServerService.java中我们给上层框架返回一个匿名类，至于里面的onNext、onCompleted方法何时被调用是上层框架决定的，另外还准备了成员变量totalCount，这样就可以记录总数了，由于请求参数是流，因此匿名类的onNext会被多次调用，并且由于返回值是流，因此onNext中调用了responseObserver.onNext方法来响应流中的每个请求，这样客户端就不断收到服务端的响应数据(即客户端的onNext方法会被多次调用)：

package com.example.localserver.server;

import com.example.springbootgrpclib.grpc.protobuf.MyRequest;
import com.example.springbootgrpclib.grpc.protobuf.MyResponse;
import com.example.springbootgrpclib.grpc.protobuf.SimpleGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;

@GrpcService
@Slf4j
public class GrpcServerService extends SimpleGrpc.SimpleImplBase {

    //双向流
    @Override
    public StreamObserver manyToMany(StreamObserver responseObserver) {
        // 返回匿名类，给上层框架使用
        return new StreamObserver() {

            // 记录总量
            private int totalCount = 0;

            @Override
            public void onNext(MyRequest myRequest) {
                log.info("正在处理[{}]，数量为[{}]",
                        myRequest.getName(),
                        myRequest.getValue());

                // 增加总量
                totalCount += myRequest.getValue();

                Integer value;
                String message;
                if (0 == myRequest.getValue() % 2) {
                    value = myRequest.getValue() / 2;
                    message = String.format(myRequest.getName() + "的值是2的倍数，减半[%d]成功", value);
                } else {
                    value = (myRequest.getValue() + 1) / 2;
                    message = String.format(myRequest.getName() + "的值不是2的倍数，减半[%d]成功", value);
                }

                responseObserver.onNext(MyResponse.newBuilder()
                        .setResult(value)
                        .setMessage(message)
                        .build());
            }

            @Override
            public void onError(Throwable t) {
                log.error("异常", t);
            }

            @Override
            public void onCompleted() {
                log.info("共计[{}]", totalCount);
                responseObserver.onCompleted();
            }
        };
    }
}

gRPC客户端代码

package com.example.localclient.service;

import com.example.springbootgrpclib.grpc.protobuf.MyRequest;
import com.example.springbootgrpclib.grpc.protobuf.MyResponse;
import com.example.springbootgrpclib.grpc.protobuf.SimpleGrpc;
import io.grpc.StatusRuntimeException;
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.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class GrpcClientService {

    @GrpcClient("local-grpc-server")
    private SimpleGrpc.SimpleStub stub;

    public List manyToMany(String name,Integer value){
        CountDownLatch countDownLatch = new CountDownLatch(1);
        // responseObserver的onNext和onCompleted会在另一个线程中被执行，
        // ExtendResponseObserver继承自StreamObserver
        ExtendResponseObserver responseObserver = new ExtendResponseObserver() {

            //保存所有来自服务端的响应

            private List strings = new ArrayList<>();
            @Override
            public List getExtra() {
                return strings;
            }


            @Override
            public void onNext(MyResponse myResponse) {
                log.info("on next");
                // 放入匿名类的成员变量中
                strings.add(String.format("值[%d]，返回信息:%s" , myResponse.getResult(), myResponse.getMessage()));
            }

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

            @Override
            public void onCompleted() {
                log.info("on complete");
                // 执行了countDown方法后，前面执行countDownLatch.await方法的线程就不再wait了，
                // 会继续往下执行
                countDownLatch.countDown();
            }
        };

        // 远程调用，此时数据还没有给到服务端
        StreamObserver requestObserver = stub.manyToMany(responseObserver);

        for(int i=1; i <= 10; i++) {
            // 发送一笔数据到服务端
            requestObserver.onNext(MyRequest.newBuilder()
                    .setName(name)
                    .setValue(value + i)
                    .build());
        }

        // 客户端告诉服务端：数据已经发完了
        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 (List) responseObserver.getExtra();
    }

}

Java版gRPC的使用之二：服务端流、客户端流和双向流

Go语言相关栏目本月热门文章