聊一聊 gRPC 的四种通信模式

江南一点雨 2023-03-28 原文

前面一篇文章松哥和大家聊了 gRPC 的基本用法，今天我们再来稍微深入一点点，来看下 gRPC 中四种不同的通信模式。

gRPC 中四种不同的通信模式分别是：

一元 RPC
服务端流 RPC
客户端流 RPC
双向流 RPC

接下来松哥就通过四个完整的案例，来分别和向伙伴们演示这四种不同的通信模式。

1. 准备工作

关于 gRPC 的基础知识我们就不啰嗦了，咱们直接来看我今天的 proto 文件，如下：

这次我新建了一个名为 book.proto 的文件，这里主要定义了一些图书相关的方法，如下：

syntax = "proto3";

option java_multiple_files = true;
option java_package = "org.javaboy.grpc.demo";
option java_outer_classname = "BookServiceProto";
import "google/protobuf/wrappers.proto";

package book;

service BookService {
  rpc addBook(Book) returns (google.protobuf.StringValue);
  rpc getBook(google.protobuf.StringValue) returns (Book);
  rpc searchBooks(google.protobuf.StringValue) returns (stream Book);
  rpc updateBooks(stream Book) returns (google.protobuf.StringValue);
  rpc processBooks(stream google.protobuf.StringValue) returns (stream BookSet);
}

message Book {
  string id = 1;
  repeated string tags = 2;
  string name = 3;
  float price = 4;
  string author = 5;
}

message BookSet {
  string id = 1;
  repeated Book bookList = 3;
}

这个文件中，有一些内容我们在上篇文章中都讲过了，讲过的我就不再重复了，我说一些上篇文章没有涉及到的东西：

由于我们在这个文件中，引用了 Google 提供的 StringValue（google.protobuf.StringValue），所以这个文件上面我们首先用 import 导入相关的文件，导入之后，才可以使用。
在方法参数和返回值中出现的 stream，就表示这个方法的参数或者返回值是流的形式（其实就是数据可以多次传输）。
message 中出现了一个上篇文章没有的关键字 repeated，这个表示这个字段可以重复，可以简单理解为这就是我们 Java 中的数组。

好了，和上篇文章相比，本文主要就是这几个地方不一样。

proto 文件写好之后，按照上篇文章介绍的方法进行编译，生成对应的代码，这里就不再重复了。

2. 一元 RPC

一元 RPC 是一种比较简单的 RPC 模式，其实说白了我们上篇文章和大家介绍的就是一种一元 RPC，也就是客户端发起一个请求，服务端给出一个响应，然后请求结束。

上面我们定义的五个方法中，addBook 和 getBook 都算是一种一元 RPC。

2.1 addBook

先来看 addBook 方法，这个方法的逻辑很简单，我们提前在服务端准备一个 Map 用来保存 Book，addBook 调用的时候，就把 book 对象存入到 Map 中，并且将 book 的 ID 返回，大家就这样一件事，来看看服务端的代码：

public class BookServiceImpl extends BookServiceGrpc.BookServiceImplBase {
    private Map<String, Book> bookMap = new HashMap<>();

    public BookServiceImpl() {
        Book b1 = Book.newBuilder().setId("1").setName("三国演义").setAuthor("罗贯中").setPrice(30).addTags("明清小说").addTags("通俗小说").build();
        Book b2 = Book.newBuilder().setId("2").setName("西游记").setAuthor("吴承恩").setPrice(40).addTags("志怪小说").addTags("通俗小说").build();
        Book b3 = Book.newBuilder().setId("3").setName("水浒传").setAuthor("施耐庵").setPrice(50).addTags("明清小说").addTags("通俗小说").build();
        bookMap.put("1", b1);
        bookMap.put("2", b2);
        bookMap.put("3", b3);
    }

    @Override
    public void addBook(Book request, StreamObserver<StringValue> responseObserver) {
        bookMap.put(request.getId(), request);
        responseObserver.onNext(StringValue.newBuilder().setValue(request.getId()).build());
        responseObserver.onCompleted();
    }
}

看过上篇文章的小伙伴，我觉得这段代码应该很好理解。

客户端调用方式如下：

public class BookServiceClient {
    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        BookServiceGrpc.BookServiceStub stub = BookServiceGrpc.newStub(channel);
        addBook(stub);
    }

    private static void addBook(BookServiceGrpc.BookServiceStub stub) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        stub.addBook(Book.newBuilder().setPrice(99).setId("100").setName("java").setAuthor("javaboy").build(), new StreamObserver<StringValue>() {
            @Override
            public void onNext(StringValue stringValue) {
                System.out.println("stringValue.getValue() = " + stringValue.getValue());
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                countDownLatch.countDown();
                System.out.println("添加完毕");
            }
        });
        countDownLatch.await();
    }
}

这里我使用了 CountDownLatch 来实现线程等待，等服务端给出响应之后，客户端再结束。这里在回调的 onNext 方法中，我们就可以拿到服务端的返回值。

2.2 getBook

getBook 跟上面的 addBook 类似，先来看服务端代码，如下：

public class BookServiceImpl extends BookServiceGrpc.BookServiceImplBase {
    private Map<String, Book> bookMap = new HashMap<>();

    public BookServiceImpl() {
        Book b1 = Book.newBuilder().setId("1").setName("三国演义").setAuthor("罗贯中").setPrice(30).addTags("明清小说").addTags("通俗小说").build();
        Book b2 = Book.newBuilder().setId("2").setName("西游记").setAuthor("吴承恩").setPrice(40).addTags("志怪小说").addTags("通俗小说").build();
        Book b3 = Book.newBuilder().setId("3").setName("水浒传").setAuthor("施耐庵").setPrice(50).addTags("明清小说").addTags("通俗小说").build();
        bookMap.put("1", b1);
        bookMap.put("2", b2);
        bookMap.put("3", b3);
    }

    @Override
    public void getBook(StringValue request, StreamObserver<Book> responseObserver) {
        String id = request.getValue();
        Book book = bookMap.get(id);
        if (book != null) {
            responseObserver.onNext(book);
            responseObserver.onCompleted();
        } else {
            responseObserver.onCompleted();
        }
    }
}

这个 getBook 就是根据客户端传来的 id，从 Map 中查询到一个 Book 并返回。

客户端调用代码如下：

public class BookServiceClient {
    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        BookServiceGrpc.BookServiceStub stub = BookServiceGrpc.newStub(channel);
        getBook(stub);
    }

    private static void getBook(BookServiceGrpc.BookServiceStub stub) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        stub.getBook(StringValue.newBuilder().setValue("2").build(), new StreamObserver<Book>() {
            @Override
            public void onNext(Book book) {
                System.out.println("book = " + book);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                countDownLatch.countDown();
                System.out.println("查询完毕");
            }
        });
        countDownLatch.await();
    }
}

小伙伴们大概也能看出来，addBook 和 getBook 基本上操作套路是一模一样的。

3. 服务端流 RPC

前面的一元 RPC，客户端发起一个请求，服务端给出一个响应，请求就结束了。服务端流则是客户端发起一个请求，服务端给一个响应序列，这个响应序列组成一个流。

上面我们给出的 searchBook 就是这样一个例子，searchBook 是传递图书的 tags 参数，然后在服务端查询哪些书的 tags 满足条件，将满足条件的书全部都返回去。

我们来看下服务端的代码：

public class BookServiceImpl extends BookServiceGrpc.BookServiceImplBase {
    private Map<String, Book> bookMap = new HashMap<>();

    public BookServiceImpl() {
        Book b1 = Book.newBuilder().setId("1").setName("三国演义").setAuthor("罗贯中").setPrice(30).addTags("明清小说").addTags("通俗小说").build();
        Book b2 = Book.newBuilder().setId("2").setName("西游记").setAuthor("吴承恩").setPrice(40).addTags("志怪小说").addTags("通俗小说").build();
        Book b3 = Book.newBuilder().setId("3").setName("水浒传").setAuthor("施耐庵").setPrice(50).addTags("明清小说").addTags("通俗小说").build();
        bookMap.put("1", b1);
        bookMap.put("2", b2);
        bookMap.put("3", b3);
    }
    @Override
    public void searchBooks(StringValue request, StreamObserver<Book> responseObserver) {
        Set<String> keySet = bookMap.keySet();
        String tags = request.getValue();
        for (String key : keySet) {
            Book book = bookMap.get(key);
            int tagsCount = book.getTagsCount();
            for (int i = 0; i < tagsCount; i++) {
                String t = book.getTags(i);
                if (t.equals(tags)) {
                    responseObserver.onNext(book);
                    break;
                }
            }
        }
        responseObserver.onCompleted();
    }
}

小伙伴们看下，这段 Java 代码应该很好理解：

首先从 request 中提取客户端传来的 tags 参数。
遍历 bookMap，查看每一本书的 tags 是否等于客户端传来的 tags，如果相等，说明添加匹配，则通过 responseObserver.onNext(book); 将这本书写回到客户端。
等所有操作都完成后，执行 responseObserver.onCompleted();，表示服务端的响应序列结束了，这样客户端也就知道请求结束了。

我们来看看客户端的代码，如下：

public class BookServiceClient {
    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        BookServiceGrpc.BookServiceStub stub = BookServiceGrpc.newStub(channel);
        searchBook(stub);
    }
    private static void searchBook(BookServiceGrpc.BookServiceStub stub) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        stub.searchBooks(StringValue.newBuilder().setValue("明清小说").build(), new StreamObserver<Book>() {
            @Override
            public void onNext(Book book) {
                System.out.println(book);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                countDownLatch.countDown();
                System.out.println("查询完毕！");
            }
        });
        countDownLatch.await();
    }
}

客户端的代码好理解，搜索的关键字是明清小说，每当服务端返回一次数据的时候，客户端回调的 onNext 方法就会被触发一次，当服务端之行了 responseObserver.onCompleted(); 之后，客户端的 onCompleted 方法也会被触发。

这个就是服务端流，客户端发起一个请求，服务端通过 onNext 可以多次写回数据。

4. 客户端流 RPC

客户端流则是客户端发起多个请求，服务端只给出一个响应。

上面的 updateBooks 就是一个客户端流的案例，客户端想要修改图书，可以发起多个请求修改多本书，服务端则收集多次修改的结果，将之汇总然后一次性返回给客户端。

我们先来看看服务端的代码：

public class BookServiceImpl extends BookServiceGrpc.BookServiceImplBase {
    private Map<String, Book> bookMap = new HashMap<>();

    public BookServiceImpl() {
        Book b1 = Book.newBuilder().setId("1").setName("三国演义").setAuthor("罗贯中").setPrice(30).addTags("明清小说").addTags("通俗小说").build();
        Book b2 = Book.newBuilder().setId("2").setName("西游记").setAuthor("吴承恩").setPrice(40).addTags("志怪小说").addTags("通俗小说").build();
        Book b3 = Book.newBuilder().setId("3").setName("水浒传").setAuthor("施耐庵").setPrice(50).addTags("明清小说").addTags("通俗小说").build();
        bookMap.put("1", b1);
        bookMap.put("2", b2);
        bookMap.put("3", b3);
    }
    
    @Override
    public StreamObserver<Book> updateBooks(StreamObserver<StringValue> responseObserver) {
        StringBuilder sb = new StringBuilder("更新的图书 ID 为：");
        return new StreamObserver<Book>() {
            @Override
            public void onNext(Book book) {
                bookMap.put(book.getId(), book);
                sb.append(book.getId())
                        .append(",");
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                responseObserver.onNext(StringValue.newBuilder().setValue(sb.toString()).build());
                responseObserver.onCompleted();
            }
        };
    }
}

客户端每发送一本书来，就会触发服务端的 onNext 方法，然后我们在这方法中进行图书的更新操作，并记录更新结果。最后，我们在 onCompleted 方法中，将更新结果汇总返回给客户端，基本上就是这样一个流程。

我们再来看看客户端的代码：

public class BookServiceClient {
    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        BookServiceGrpc.BookServiceStub stub = BookServiceGrpc.newStub(channel);
        updateBook(stub);
    }

    private static void updateBook(BookServiceGrpc.BookServiceStub stub) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        StreamObserver<Book> request = stub.updateBooks(new StreamObserver<StringValue>() {
            @Override
            public void onNext(StringValue stringValue) {
                System.out.println("stringValue.getValue() = " + stringValue.getValue());
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                System.out.println("更新完毕");
                countDownLatch.countDown();
            }
        });
        request.onNext(Book.newBuilder().setId("1").setName("a").setAuthor("b").build());
        request.onNext(Book.newBuilder().setId("2").setName("c").setAuthor("d").build());
        request.onCompleted();
        countDownLatch.await();
    }
}

在客户端这块，updateBooks 方法会返回一个 StreamObserver对象，调用该对象的 onNext 方法就是给服务端传递数据了，可以传递多个数据，调用该对象的 onCompleted 方法就是告诉服务端数据传递结束了，此时也会触发服务端的 onCompleted 方法，服务端的 onCompleted 方法执行之后，进而触发了客户端的 onCompleted 方法。

5. 双向流 RPC

双向流其实就是 3、4 小节的合体。即客户端多次发送数据，服务端也多次响应数据。

我们先来看下服务端的代码：

public class BookServiceImpl extends BookServiceGrpc.BookServiceImplBase {
    private Map<String, Book> bookMap = new HashMap<>();
    private List<Book> books = new ArrayList<>();

    public BookServiceImpl() {
        Book b1 = Book.newBuilder().setId("1").setName("三国演义").setAuthor("罗贯中").setPrice(30).addTags("明清小说").addTags("通俗小说").build();
        Book b2 = Book.newBuilder().setId("2").setName("西游记").setAuthor("吴承恩").setPrice(40).addTags("志怪小说").addTags("通俗小说").build();
        Book b3 = Book.newBuilder().setId("3").setName("水浒传").setAuthor("施耐庵").setPrice(50).addTags("明清小说").addTags("通俗小说").build();
        bookMap.put("1", b1);
        bookMap.put("2", b2);
        bookMap.put("3", b3);
    }
    @Override
    public StreamObserver<StringValue> processBooks(StreamObserver<BookSet> responseObserver) {
        return new StreamObserver<StringValue>() {
            @Override
            public void onNext(StringValue stringValue) {
                Book b = Book.newBuilder().setId(stringValue.getValue()).build();
                books.add(b);
                if (books.size() == 3) {
                    BookSet bookSet = BookSet.newBuilder().addAllBookList(books).build();
                    responseObserver.onNext(bookSet);
                    books.clear();
                }
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                BookSet bookSet = BookSet.newBuilder().addAllBookList(books).build();
                responseObserver.onNext(bookSet);
                books.clear();
                responseObserver.onCompleted();
            }
        };
    }
}

这段代码没有实际意义，单纯为了给小伙伴们演示双向流，我的操作逻辑是客户端传递多个 ID 到服务端，然后服务端根据这些 ID 构建对应的 Book 对象，然后三个三个一组，再返回给客户端。客户端每次发送一个请求，都会触发服务端的 onNext 方法，我们在这个方法中对请求分组返回。最后如果还有剩余的请求，我们在 onCompleted() 方法中返回。

再来看看客户端的代码：

public class BookServiceClient {
    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        BookServiceGrpc.BookServiceStub stub = BookServiceGrpc.newStub(channel);
        processBook(stub);
    }

    private static void processBook(BookServiceGrpc.BookServiceStub stub) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        StreamObserver<StringValue> request = stub.processBooks(new StreamObserver<BookSet>() {
            @Override
            public void onNext(BookSet bookSet) {
                System.out.println("bookSet = " + bookSet);
                System.out.println("=============");
            }

            @Override
            public void onError(Throwable throwable) {
            }

            @Override
            public void onCompleted() {
                System.out.println("处理完毕！");
                countDownLatch.countDown();
            }
        });
        request.onNext(StringValue.newBuilder().setValue("a").build());
        request.onNext(StringValue.newBuilder().setValue("b").build());
        request.onNext(StringValue.newBuilder().setValue("c").build());
        request.onNext(StringValue.newBuilder().setValue("d").build());
        request.onCompleted();
        countDownLatch.await();
    }
}

这个客户端的代码跟第四小节一模一样，不再赘述了。

好啦，这就是松哥和小伙伴们介绍的 gRPC 的四种不同的通信模式，文章中只给出了一些关键代码，如果小伙伴们没看明白，建议结合上篇文章一起阅读就懂啦～

gRPC 聊一 span color style 网络网络管理 $gRPC 通信模式

有关聊一聊 gRPC 的四种通信模式的更多相关文章

ruby-on-rails - Rails - 子类化模型的设计模式是什么？ - 2
我有一个模型:classItem项目有一个属性“商店”基于存储的值，我希望Item对象对特定方法具有不同的行为。Rails中是否有针对此的通用设计模式？如果方法中没有大的if-else语句，这是如何干净利落地完成的？最佳答案通常通过Single-TableInheritance. 关于ruby-on-rails-Rails-子类化模型的设计模式是什么？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.co
ruby - 解析 RDFa、微数据等的最佳方式是什么，使用统一的模式/词汇(例如 schema.org)存储和显示信息 - 2
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
ruby-on-rails - Rails 应用程序之间的通信 - 2
我构建了两个需要相互通信和发送文件的Rails应用程序。例如，一个Rails应用程序会发送请求以查看其他应用程序数据库中的表。然后另一个应用程序将呈现该表的json并将其发回。我还希望一个应用程序将存储在其公共(public)目录中的文本文件发送到另一个应用程序的公共(public)目录。我从来没有做过这样的事情，所以我什至不知道从哪里开始。任何帮助，将不胜感激。谢谢! 最佳答案无论Rails是什么，几乎所有Web应用程序都有您的要求，大多数现代Web应用程序都需要相互通信。但是有一个小小的理解需要你坚持下去，网站不应直接访问彼此
ruby - 如何在续集中重新加载表模式？ - 2
鉴于我有以下迁移:Sequel.migrationdoupdoalter_table:usersdoadd_column:is_admin,:default=>falseend#SequelrunsaDESCRIBEtablestatement,whenthemodelisloaded.#Atthispoint,itdoesnotknowthatusershaveais_adminflag.#Soitfails.@user=User.find(:email=>"admin@fancy-startup.example")@user.is_admin=true@user.save!ende
ruby - 是否有用于序列化和反序列化各种格式的对象层次结构的模式？ - 2
给定一个复杂的对象层次结构，幸运的是它不包含循环引用，我如何实现支持各种格式的序列化？我不是来讨论实际实现的。相反，我正在寻找可能会派上用场的设计模式提示。更准确地说:我正在使用Ruby，我想解析XML和JSON数据以构建复杂的对象层次结构。此外，应该可以将该层次结构序列化为JSON、XML和可能的HTML。我可以为此使用Builder模式吗？在任何提到的情况下，我都有某种结构化数据-无论是在内存中还是文本中-我想用它来构建其他东西。我认为将序列化逻辑与实际业务逻辑分开会很好，这样我以后就可以轻松支持多种XML格式。最佳答案我最
MIMO-OFDM无线通信技术及MATLAB实现（1）无线信道：传播和衰落 - 2
MIMO技术的优缺点优点通过下面三个增益来总体概括：阵列增益。阵列增益是指由于接收机通过对接收信号的相干合并而活得的平均SNR的提高。在发射机不知道信道信息的情况下，MIMO系统可以获得的阵列增益与接收天线数成正比复用增益。在采用空间复用方案的MIMO系统中，可以获得复用增益，即信道容量成倍增加。信道容量的增加与min(Nt,Nr)成正比分集增益。在采用空间分集方案的MIMO系统中，可以获得分集增益，即可靠性性能的改善。分集增益用独立衰落支路数来描述，即分集指数。在使用了空时编码的MIMO系统中，由于接收天线或发射天线之间的间距较远，可认为它们各自的大尺度衰落是相互独立的，因此分布式MIMO
ruby-on-rails - environment.rb 中设置的常量在开发模式中消失 - 2
了解Rails缓存如何工作的人可以真正帮助我。这是嵌套在Rails::Initializer.runblock中的代码:config.after_initializedoSomeClass.const_set'SOME_CONST','SOME_VAL'end现在，如果我运行script/server并发出请求，一切都很好。然而，在我的Rails应用程序的第二个请求中，一切都因单元化常量错误而变得糟糕。在生产模式下，我可以成功发出第二个请求，这意味着常量仍然存在。我已通过将以上内容更改为以下内容来解决问题:config.after_initializedorequire'some_cl
Ruby:标准递归模式 - 2
我经常迷上ruby的一件事是递归模式。例如，假设我有一个数组，它可能包含无限深度的数组作为元素。所以，例如:my_array=[1,[2,3,[4,5,[6,7]]]]我想创建一个方法，可以将数组展平为[1,2,3,4,5,6,7]。我知道.flatten可以完成这项工作，但这个问题是作为我经常遇到的递归问题的一个例子-因此我试图找到一个更可重用的解决方案。简而言之-我猜这种事情有一个标准模式，但我想不出任何特别优雅的东西。任何想法表示赞赏最佳答案递归是一种方法，它不依赖于语言。您在编写算法时要考虑两种情况:再次调用函数的情
ruby - 在 Ruby 中查找多个正则表达式匹配的模式和位置 - 2
这应该是一个简单的问题，但我找不到任何相关信息。给定一个Ruby中的正则表达式，对于每个匹配项，我需要检索匹配的模式$1、$2，但我还需要匹配位置。我知道=~运算符为我提供了第一个匹配项的位置，而string.scan(/regex/)为我提供了所有匹配模式。如果可能，我需要在同一步骤中获得两个结果。最佳答案 MatchDatastring.scan(regex)do$1#Patternatfirstposition$2#Patternatsecondposition$~.offset(1)#Startingandendingpo
ruby - sinatra 框架的 MVC 模式 - 2
我想开始使用“Sinatra”框架进行编码，但我找不到该框架的“MVC”模式。是“MVC-Sinatra”模式或框架吗？最佳答案您可能想查看Padrino这是一个围绕Sinatra构建的框架，可为您的项目提供更“类似Rails”的感觉，但没有那么多隐藏的魔法。这是使用Sinatra可以做什么的一个很好的例子。虽然如果您需要开始使用这很好，但我个人建议您将它用作学习工具，以对您来说最有意义的方式使用Sinatra构建您自己的应用程序。写一些测试/期望，写一些代码，通过测试-重复:)至于ORM，你还应该结帐Sequel其中(imho