当前位置: 开发笔记 > 编程语言 > 正文

gRPC的通信方式客户端流式、服务端流式、双向流式在Java的调用示例

作者：维他柠檬鱼1949 | 来源：互联网 | 2023-05-16 12:56

gPRC简介以及Java中使用gPRC实现客户端与服务端通信(附代码下载)：https:blog.csdn.netBADAO_LIUMANG_QIZHIarticledetails

gPRC简介以及Java中使用gPRC实现客户端与服务端通信(附代码下载)：

https://blog.csdn.net/BADAO_LIUMANG_QIZHI/article/details/108711541

在上面的博客中介绍了gRPC以及使用最基本的rpc通信方式即一个请求对象返回一个响应的方式进行通信。

除此之外gRPC还有以下三种方式。

服务端流式

一个请求对象,服务端返回多个结果对象

proto示例语法

rpc GetStudentsByAge(StudentRequest) returns (stream StudentResponse) {}

客户端流式

客户端传入多个请求对象，服务端返回一个响应结果。

proto示例语法

rpc GetStudentsWrapperByAges(stream StudentRequest) returns (StudentResponseList) {}

双向流式

传入多个对象可以返回多个响应对象

注：

博客：
https://blog.csdn.net/badao_liumang_qizhi
关注公众号
霸道的程序猿
获取编程相关电子书、教程推送与免费下载。

实现

服务端流式实现

在上面博客的基础上，打开Person.proto文件

message StudentRequest {
    int32 age = 1;
}

message StudentResponse {
    string name = 1;
    int32 age = 2;
    string city = 3;
}

添加两个message作为请求和响应对象。

因为gRPC的请求和响应对象必须在message中定义，不能直接使用string或者int32这种作为参数。

然后在新建接口方法

service PersonService {
    rpc GetStudentsByAge(StudentRequest) returns (stream StudentResponse) {}
}

此方法是要请求参数为一个age，然后返回多个学生对象。

然后调用插件生成代码。

然后来到PersonServiceImpl中对接口方法进行实现

    @Override
    public void getStudentsByAge(StudentRequest request, StreamObserver responseObserver) {
        System.out.println("接收到的客户端消息为:"+request.getAge());

        responseObserver.onNext(StudentResponse.newBuilder().setName("1公众号:霸道的程序猿")
                .setAge(30)
                .setCity("北京")
                .build());

        responseObserver.onNext(StudentResponse.newBuilder().setName("2公众号:霸道的程序猿")
                .setAge(40)
                .setCity("上海")
                .build());

        responseObserver.onNext(StudentResponse.newBuilder().setName("3公众号:霸道的程序猿")
                .setAge(50)
                .setCity("广州")
                .build());
        responseObserver.onCompleted();
    }

然后来到客户端中

       ManagedChannel managedChannel  = ManagedChannelBuilder.forAddress("localhost",8899)
                .usePlaintext().build();
        PersonServiceGrpc.PersonServiceBlockingStub blockingStub = PersonServiceGrpc.newBlockingStub(managedChannel);

        System.out.println("请求-流式响应，调用getRealNameByUsername");
        Iterator iter = blockingStub.getStudentsByAge(StudentRequest.newBuilder().setAge(20).build());
        while (iter.hasNext())
        {
            StudentResponse studentResponse = iter.next();
            System.out.println(studentResponse.getName());
            System.out.println(studentResponse.getAge());
            System.out.println(studentResponse.getCity());
        }

然后运行服务端后再运行客户端

gRPC的通信方式-客户端流式、服务端流式、双向流式在Java的调用示例

此时服务端

gRPC的通信方式-客户端流式、服务端流式、双向流式在Java的调用示例

客户端流式实现

打开proto文件

message StudentRequest {
    int32 age = 1;
}

message StudentResponse {
    string name = 1;
    int32 age = 2;
    string city = 3;
}

message StudentResponseList {
    repeated StudentResponse studentResponse = 1;
}

添加响应的list，要实现客户端发动流式多个请求参数(年龄)，服务端返回单个list对象，其中每个List的数据是学生对象。

添加接口方法

service PersonService {
    rpc GetStudentsWrapperByAges(stream StudentRequest) returns (StudentResponseList) {}
}

然后调用插件生成代码。打开PersonServiceImpl进行方法的实现

   @Override
    public StreamObserver getStudentsWrapperByAges(final StreamObserver responseObserver) {
        return new StreamObserver() {
            public void onNext(StudentRequest studentRequest) {
                System.out.println("onNext:"+studentRequest.getAge());
            }

            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }

            public void onCompleted() {
                StudentResponse studentResponse = StudentResponse.newBuilder()
                        .setName("公众号:霸道的程序猿")
                        .setAge(20)
                        .setCity("北京").build();
                StudentResponse studentResponse1 = StudentResponse.newBuilder()
                        .setName("1公众号:霸道的程序猿")
                        .setAge(30)
                        .setCity("上海").build();

                StudentResponseList studentResponseList = StudentResponseList.newBuilder()
                        .addStudentResponse(studentResponse).addStudentResponse(studentResponse1).build();

                responseObserver.onNext(studentResponseList);
                responseObserver.onCompleted();

            }
        };
    }

与上面不同，客户端如果是流式请求的话，那么客户端必须使用异步的stub

PersonServiceGrpc.PersonServiceStub stub = PersonServiceGrpc.newStub(managedChannel);

客户端代码为

        ManagedChannel managedChannel  = ManagedChannelBuilder.forAddress("localhost",8899)
                .usePlaintext().build();

        PersonServiceGrpc.PersonServiceStub stub = PersonServiceGrpc.newStub(managedChannel);
  
        System.out.println("-----------------------------");
        System.out.println("流式请求-响应，调用GetStudentsWrapperByAges");
        StreamObserver studentRespOnseListStreamObserver= new StreamObserver() {
            public void onNext(StudentResponseList studentResponseList) {
                studentResponseList.getStudentResponseList().forEach(studengResponse ->{
                    System.out.println(studengResponse.getName());
                    System.out.println(studengResponse.getAge());
                    System.out.println(studengResponse.getCity());
                });
            }

            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }

            public void onCompleted() {
                System.out.println("completed");
            }
        };
        StreamObserver studentRequestStreamObserver = stub.getStudentsWrapperByAges(studentResponseListStreamObserver);
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(20).build());
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(30).build());
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(40).build());
        studentRequestStreamObserver.onCompleted();

        try {
            Thread.sleep(50000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

因为是异步的所以必须使进程进行休眠才能看到效果

运行服务端后运行客户端

gRPC的通信方式-客户端流式、服务端流式、双向流式在Java的调用示例

此时服务端

gRPC的通信方式-客户端流式、服务端流式、双向流式在Java的调用示例

双向流式实现

打开proto文件

message StreamRequest {
    string request_info = 1;
}

message StreamResponse {
    string response_info = 1;
}

新建流式请求与响应参数，然后新建接口方法

service PersonService {
    rpc BiTalk(stream StreamRequest) returns (stream StreamResponse) {}
}

然后实现接口方法

    @Override
    public StreamObserver biTalk(StreamObserver responseObserver) {
        return new StreamObserver() {
            @Override
            public void onNext(StreamRequest streamRequest) {
                System.out.println(streamRequest.getRequestInfo());
                responseObserver.onNext(StreamResponse.newBuilder().setResponseInfo(UUID.randomUUID().toString()).build());
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }

            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }

在客户端中

package com.badao.grpcjava;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.time.LocalDate;
import java.util.Iterator;

public class GrpcClient {
    public static void main(String[] args) {
        ManagedChannel managedChannel  = ManagedChannelBuilder.forAddress("localhost",8899)
                .usePlaintext().build();
        PersonServiceGrpc.PersonServiceStub stub = PersonServiceGrpc.newStub(managedChannel);

        System.out.println("-----------------------------");
        System.out.println("流式请求-流式响应，调用BiTalk");

        StreamObserver requestStreamObserver = stub.biTalk(new StreamObserver() {
            @Override
            public void onNext(StreamResponse streamResponse) {
                System.out.println(streamResponse.getResponseInfo());
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println("onComplated");
            }
        });


        for(int i =0;i<10;i++)
        {
            requestStreamObserver.onNext(StreamRequest.newBuilder().setRequestInfo(LocalDate.now().toString()).build());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }


        try {
            Thread.sleep(50000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

运行服务端后运行客户端

gRPC的通信方式-客户端流式、服务端流式、双向流式在Java的调用示例

示例代码下载

https://download.csdn.net/download/BADAO_LIUMANG_QIZHI/12883063

推荐阅读

go
Android系统源码分析Zygote和SystemServer启动过程详解

本文详细解析了Android系统源码中Zygote和SystemServer的启动过程。首先介绍了系统framework层启动的内容，帮助理解四大组件的启动和管理过程。接着介绍了AMS、PMS等系统服务的作用和调用方式。然后详细分析了Zygote的启动过程，解释了Zygote在Android启动过程中的决定作用。最后通过时序图展示了整个过程。 ... [详细]

蜡笔小新 2023-12-11 17:46:46
default
Java太阳系小游戏分析和源码详解

本文介绍了一个基于Java的太阳系小游戏的分析和源码详解。通过对面向对象的知识的学习和实践，作者实现了太阳系各行星绕太阳转的效果。文章详细介绍了游戏的设计思路和源码结构，包括工具类、常量、图片加载、面板等。通过这个小游戏的制作，读者可以巩固和应用所学的知识，如类的继承、方法的重载与重写、多态和封装等。 ... [详细]

蜡笔小新 2023-12-14 19:53:34
go
在类中定义数组时出错 - Error on defining arrays in class

Iamtryingtomakeaclassthatwillreadatextfileofnamesintoanarray,thenreturnthatarra ... [详细]

蜡笔小新 2023-12-14 17:38:12
default
SpringBoot uri统一权限管理的实现方法及步骤详解

本文详细介绍了SpringBoot中实现uri统一权限管理的方法，包括表结构定义、自动统计URI并自动删除脏数据、程序启动加载等步骤。通过该方法可以提高系统的安全性，实现对系统任意接口的权限拦截验证。 ... [详细]

蜡笔小新 2023-12-14 16:11:30
list
Mac OS 升级到11.2.2 Eclipse打不开了，报错Failed to create the Java Virtual Machine

本文介绍了在Mac OS升级到11.2.2版本后，使用Eclipse打开时出现报错Failed to create the Java Virtual Machine的问题，并提供了解决方法。 ... [详细]

蜡笔小新 2023-12-14 12:01:13
go
Hyperledger Fabric外部链码构建与运行的开发笔记

本文介绍了Hyperledger Fabric外部链码构建与运行的相关知识，包括在Hyperledger Fabric 2.0版本之前链码构建和运行的困难性，外部构建模式的实现原理以及外部构建和运行API的使用方法。通过本文的介绍，读者可以了解到如何利用外部构建和运行的方式来实现链码的构建和运行，并且不再受限于特定的语言和部署环境。 ... [详细]

蜡笔小新 2023-12-13 21:47:39
instance
iOS数据库Sqlite的SQL语句分类和常见约束关键字

本文介绍了iOS数据库Sqlite的SQL语句分类和常见约束关键字。SQL语句分为DDL、DML和DQL三种类型，其中DDL语句用于定义、删除和修改数据表，关键字包括create、drop和alter。常见约束关键字包括if not exists、if exists、primary key、autoincrement、not null和default。此外，还介绍了常见的数据库数据类型，包括integer、text和real。 ... [详细]

蜡笔小新 2023-12-12 18:42:03
list
Java中包装类的设计原因以及操作方法

本文主要介绍了Java中设计包装类的原因以及操作方法。在Java中，除了对象类型，还有八大基本类型，为了将基本类型转换成对象，Java引入了包装类。文章通过介绍包装类的定义和实现，解答了为什么需要包装类的问题，并提供了简单易用的操作方法。通过本文的学习，读者可以更好地理解和应用Java中的包装类。 ... [详细]

蜡笔小新 2023-12-12 15:48:10
go
Week04面向对象设计与继承学习总结及作业要求

本文总结了Week04面向对象设计与继承的重要知识点，包括对象、类、封装性、静态属性、静态方法、重载、继承和多态等。同时，还介绍了私有构造函数在类外部无法被调用、static不能访问非静态属性以及该类实例可以共享类里的static属性等内容。此外，还提到了作业要求，包括讲述一个在网上商城购物或在班级博客进行学习的故事，并使用Markdown的加粗标记和语句块标记标注关键名词和动词。最后，还提到了参考资料中关于UML类图如何绘制的范例。 ... [详细]

蜡笔小新 2023-12-11 16:50:17
foreach
解决.net项目中未注册“microsoft.ACE.oledb.12.0”提供程序的方法

在开发.net项目中，通过microsoft.ACE.oledb读取excel文件信息时，报错“未在本地计算机上注册“microsoft.ACE.oledb.12.0”提供程序”。本文提供了解决这个问题的方法，包括错误描述和代码示例。通过注册提供程序和修改连接字符串，可以成功读取excel文件信息。 ... [详细]

蜡笔小新 2023-12-11 15:08:02
list
MVP架构模式及其在国庆技术博客中的应用

本文介绍了MVP架构模式及其在国庆技术博客中的应用。MVP架构模式是一种演变自MVC架构的新模式，其中View和Model之间的通信通过Presenter进行。相比MVC架构，MVP架构将交互逻辑放在Presenter内部，而View直接从Model中读取数据而不是通过Controller。本文还探讨了MVP架构在国庆技术博客中的具体应用。 ... [详细]

蜡笔小新 2023-12-11 11:57:58
list
如何在php文件中添加图片？

本文详细解答了如何在php文件中添加图片的问题，包括插入图片的代码、使用PHPword在载入模板中插入图片的方法，以及使用gd库生成不同类型的图像文件的示例。同时还介绍了如何生成一个正方形文件的步骤。希望对大家有所帮助。 ... [详细]

蜡笔小新 2023-12-11 10:46:38
list
PHP cURL请求体在node.js中没有定义的问题

本文讨论了在使用PHP cURL发送POST请求时，请求体在node.js中没有定义的问题。作者尝试了多种解决方案，但仍然无法解决该问题。同时提供了当前PHP代码示例。 ... [详细]

蜡笔小新 2023-12-10 10:19:23
ip
如何修改MySQL数据库密码的方法及步骤详解

本文详细介绍了四种修改MySQL数据库密码的方法，包括使用SET PASSWORD命令、mysqladmin命令、UPDATE语句直接编辑user表的步骤和操作示例。通过本文的指导，读者可以轻松掌握修改MySQL数据库密码的技巧和方法。 ... [详细]

蜡笔小新 2023-12-09 19:47:43
default
springboot启动不了_Spring Boot + MyBatis 多模块搭建教程

作者：枫本非凡来源：www.cnblogs.comorzlinp9717399.html一、前言1、创建父工程最近公司项目准备开始重构，框 ... [详细]

蜡笔小新 2023-12-09 19:23:08

维他柠檬鱼1949

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章