当前位置： 首页 > news >正文

罗湖做网站上海外贸公司招聘职位

news 2025/11/7 2:27:19
罗湖做网站,上海外贸公司招聘职位,广州网络推广建站,郴州网站制作公司1、gRPC流 从其名称可以理解，流就是持续不断的传输。有一些业务场景请求或者响应的数据量比较大，不适合使用普通的 RPC 调用通过一次请求-响应处理，一方面是考虑数据量大对请求响应时间的影响，另一方面业务场景的设计不一 定需…

1、gRPC流

从其名称可以理解，流就是持续不断的传输。有一些业务场景请求或者响应的数据量比较大，不适合使用普通的

RPC 调用通过一次请求-响应处理，一方面是考虑数据量大对请求响应时间的影响，另一方面业务场景的设计不一

定需要一次性处理完所有数据，这时就可以使用流来分批次传输数据。

HTTP/2中有两个概念，流（stream）与帧（frame），其中帧作为HTTP/2中通信的最小传输单位，通常一个请

求或响应会被分为一个或多个帧传输，流则表示已建立连接的虚拟通道，可以传输多次请求或响应。每个帧中包含

Stream Identifier，标志所属流。HTTP/2通过流与帧实现多路复用，对于相同域名的请求，通过Stream

Identifier标识可在同一个流中进行，从而减少连接开销。 而gRPC基于HTTP/2协议传输，自然而然也实现了流式

传输，其中gRPC中共有以下三种类型的流：

1、服务端流式响应

2、客户端流式请求

3、两端双向流式

本篇主要讲讲如何实现gRPC三种流式处理。

gRPC的stream只需要在service的rpc方法描述中通过 stream 关键字指定启用流特性就好了。

1.1 单向流

单向流是指客户端和服务端只有一端开启流特性，这里的单向特指发送数据的方向。

  • 当服务端开启流时，客户端和普通 RPC 调用一样通过一次请求发送数据，服务端通过流分批次响应。

  • 当客户端开启流时，客户端通过流分批次发送请求数据，服务端接完所有数据后统一响应一次。

1.1.1 服务端流

定义一个 MultiPong 方法，在服务端开启流，功能是接收到客户端的请求后响应10次 pong 消息。

ping.proto文件的编写：

// ping.proto
// 指定proto版本
syntax = "proto3";
// 指定包名
package protos;
// 指定go包路径
option go_package = "protos/ping";
// 定义PingPong服务
service PingPong {// Ping发送ping请求,接收pong响应// 服务端流模式,在响应消息前添加stream关键字rpc MultiPong(PingRequest) returns (stream PongResponse);
}// PingRequest请求结构
message PingRequest {// value字段为string类型string value = 1; 
}// PongResponse 响应结构
message PongResponse {// value字段为string类型string value = 1; 
}

ping.pb.go文件的生成：

$ protoc --go_out=plugins=grpc:. ping.proto

服务端实现，server.go的编写，第二个参数为 stream 对象的引用，可以通过它的 Send 方法发送数据。

package mainimport (// 引入编译生成的包pb "demo/protos/ping""google.golang.org/grpc""log""net"
)// PingPongServer 实现 pb.PingPongServer 接口
type PingPongServer struct {pb.UnimplementedPingPongServer
}// MultiPong 服务端流模式
func (s *PingPongServer) MultiPong(req *pb.PingRequest, stream pb.PingPong_MultiPongServer) error {for i := 0; i < 10; i++ {data := &pb.PongResponse{Value: "pong"}// 发送消息err := stream.Send(data)if err != nil {return err}}return nil
}// 启动server
func main() {srv := grpc.NewServer()// 注册 PingPongServerpb.RegisterPingPongServer(srv, &PingPongServer{})lis, err := net.Listen("tcp", ":7009")if err != nil {log.Fatalln(err)}log.Println("listen on 7009")srv.Serve(lis)
}

# 启动server
$ go run server.go
2023/02/10 20:51:04 listen on 7009

客户端实现，client.go的编写，请求方式和普通 RPC 没有区别，重点关注对响应数据流的处理，通过一个 for

循环接收数据直到结束。

package mainimport ("context"pb "demo/protos/ping" // 引入编译生成的包"google.golang.org/grpc""io""log"
)// Ping 单次请求-响应模式
func main() {conn, err := grpc.Dial("localhost:7009", grpc.WithInsecure())if err != nil {log.Fatalln(err)}defer conn.Close()// 实例化客户端并调用client := pb.NewPingPongClient(conn)// 获得对 stream 对象的引用stream, err := client.MultiPong(context.Background(), &pb.PingRequest{Value: "ping"})if err != nil {log.Fatalln(err)}// 循环接收响应数据流for {msg, err := stream.Recv()if err != nil {// 数据结束if err == io.EOF {break}log.Fatalln(err)}log.Println(msg.Value)}
}

# 客户端运行
$ go run client.go
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong
2023/02/10 20:54:34 pong

# 目录结构
$ tree demo
demo
├── client.go
├── go.mod
├── go.sum
├── ping.proto
├── protos
│   └── ping
│       └── ping.pb.go
└── server.go2 directories, 6 files

1.1.2 客户端流

定义一个 MultiPing 方法，在客户端开启流，功能是持续发送多个 ping 请求，服务端统一响应一次。

ping.proto文件的编写：

// ping.proto
// 指定proto版本
syntax = "proto3"; 
// 指定包名
package protos;     
// 指定go包路径
option go_package = "protos/ping";
// 定义PingPong服务
service PingPong {// Ping 发送 ping 请求，接收 pong 响应// 客户端流模式，在请求消息前添加 stream 关键字rpc MultiPing(stream PingRequest) returns (PongResponse);
}// PingRequest 请求结构
message PingRequest {string value = 1; // value字段为string类型
}// PongResponse 响应结构
message PongResponse {string value = 1; // value字段为string类型
}

ping.pb.go文件的生成：

$ protoc --go_out=plugins=grpc:. ping.proto

服务端实现，server.go的编写，只有一个参数为 stream 对象的引用，可以通过它的 Recv 方法接收数据。使

用 SendAndClose 方法关闭流并响应，服务端可以根据需要提前关闭。

package mainimport ("fmt"// 引入编译生成的包pb "demo/protos/ping""google.golang.org/grpc""io""log""net"
)// PingPongServer 实现 pb.PingPongServer 接口
type PingPongServer struct {pb.UnimplementedPingPongServer
}// MultiPing 客户端流模式
func (s *PingPongServer) MultiPing(stream pb.PingPong_MultiPingServer) error {msgs := []string{}for {// 提前结束接收消息if len(msgs) > 5 {return stream.SendAndClose(&pb.PongResponse{Value: "ping enough, max 5"})}msg, err := stream.Recv()if err != nil {// 客户端消息结束，返回响应信息if err == io.EOF {return stream.SendAndClose(&pb.PongResponse{Value: fmt.Sprintf("got %d ping", len(msgs))})}return err}msgs = append(msgs, msg.Value)}
}// 启动server
func main() {srv := grpc.NewServer()// 注册 PingPongServerpb.RegisterPingPongServer(srv, &PingPongServer{})lis, err := net.Listen("tcp", ":7009")if err != nil {log.Fatalln(err)}log.Println("listen on 7009")srv.Serve(lis)
}

# 启动server
$ go run server.go
2023/02/10 21:26:42 listen on 7009

客户端实现，client.go的编写，调用 MultiPing 方法时不再指定请求参数，而是通过返回的 stream 对象的

Send 分批发送数据。

package mainimport ("context"pb "demo/protos/ping" // 引入编译生成的包"google.golang.org/grpc""log"
)// Ping 单次请求-响应模式
func main() {conn, err := grpc.Dial("localhost:7009", grpc.WithInsecure())if err != nil {log.Fatalln(err)}defer conn.Close()// 实例化客户端并调用client := pb.NewPingPongClient(conn)// 获得对stream对象的引用// 调用并得到stream对象stream, err := client.MultiPing(context.Background())if err != nil {log.Fatalln(err)}// 发送数据for i := 0; i < 6; i++ {data := &pb.PingRequest{Value: "ping"}err = stream.Send(data)if err != nil {log.Fatalln(err)}}// 发送结束并获取服务端响应res, err := stream.CloseAndRecv()if err != nil {log.Fatalln(err)}log.Println(res.Value)
}

# 启动客户端
# 发送3个ping
$ go run client.go
2023/02/10 21:32:31 got 3 ping
# 发送6个ping
$ go run client.go
2023/02/10 21:32:31 ping enough, max 5

# 项目结构
$ tree demo
demo
├── client.go
├── go.mod
├── go.sum
├── ping.proto
├── protos
│   └── ping
│       └── ping.pb.go
└── server.go2 directories, 6 files

1.2 双向流

双向流是指客户端在发送数据和服务端响应数据的过程中都启用流特性，实际上单向流只是双向流的特例，有了上

面的基础，双向流就很好理解了。

定义一个 MultiPingPong 方法，在客户端和服务端都开启流，功能是服务端每接收到两个 ping 就响应一次

pong。

ping.proto编写：

// ping.proto
// 指定proto版本
syntax = "proto3"; 
// 指定包名
package protos;     
// 指定go包路径
option go_package = "protos/ping";// 定义PingPong服务
service PingPong {// Ping 发送 ping 请求，接收 pong 响应// 双向流模式rpc MultiPingPong(stream PingRequest) returns (stream PongResponse);
}// PingRequest 请求结构
message PingRequest {string value = 1; // value字段为string类型
}// PongResponse 响应结构
message PongResponse {string value = 1; // value字段为string类型
}

ping.pb.go文件的生成：

$ protoc --go_out=plugins=grpc:. ping.proto

服务端实现，server.go的编写，同样通过 stream 的 Recv 和 Send 方法接收和发送数据。

package mainimport (pb "demo/protos/ping" // 引入编译生成的包"google.golang.org/grpc""io""log""net"
)// PingPongServer 实现 pb.PingPongServer 接口
type PingPongServer struct {pb.UnimplementedPingPongServer
}func (s *PingPongServer) MultiPingPong(stream pb.PingPong_MultiPingPongServer) error {msgs := []string{}for {// 接收消息msg, err := stream.Recv()if err != nil {if err == io.EOF {break}return err}msgs = append(msgs, msg.Value)// 每收到两个消息响应一次if len(msgs)%2 == 0 {err = stream.Send(&pb.PongResponse{Value: "pong"})if err != nil {return err}}}return nil
}// 启动server
func main() {srv := grpc.NewServer()// 注册 PingPongServerpb.RegisterPingPongServer(srv, &PingPongServer{})lis, err := net.Listen("tcp", ":7009")if err != nil {log.Fatal(err)}log.Println("listen on 7009")srv.Serve(lis)
}

# 启动server
$ go run server.go
2023/02/10 21:26:42 listen on 7009

客户端实现，client.go的编写，这里在另外一个 goroutine 里处理接收数据的逻辑来演示同时发送和接收数

据。

package mainimport ("context"pb "demo/protos/ping" // 引入编译生成的包"google.golang.org/grpc""io""log""time"
)// Ping 单次请求-响应模式
func main() {conn, err := grpc.Dial("localhost:7009", grpc.WithInsecure())if err != nil {log.Fatal(err)}defer conn.Close()// 实例化客户端并调用client := pb.NewPingPongClient(conn)stream, err := client.MultiPingPong(context.Background())if err != nil {log.Fatal(err)}// 在另一个goroutine中处理接收数据c := make(chan struct{})go func(stream pb.PingPong_MultiPingPongClient, c chan struct{}) {defer func() {c <- struct{}{}}()for {msg, err := stream.Recv()if err != nil {if err == io.EOF {break}log.Fatal(err)}log.Printf("recv:%s\n", msg.Value)}}(stream, c)// 发送数据for i := 0; i < 6; i++ {data := &pb.PingRequest{Value: "ping"}err = stream.Send(data)if err != nil {log.Fatal(err)}log.Printf("send:%s\n", data.Value)// 延时一段时间发送，等待响应结果time.Sleep(500 * time.Millisecond)}// 结束发送stream.CloseSend()// 等待接收完成<-c
}

# 启动客户端
$ go run client.go
2023/02/10 21:48:26 send:ping
2023/02/10 21:48:26 send:ping
2023/02/10 21:48:26 recv:pong
2023/02/10 21:48:27 send:ping
2023/02/10 21:48:27 send:ping
2023/02/10 21:48:27 recv:pong
2023/02/10 21:48:28 send:ping
2023/02/10 21:48:28 send:ping
2023/02/10 21:48:28 recv:pong

# 项目结构
$ tree demo
demo
├── client.go
├── go.mod
├── go.sum
├── ping.proto
├── protos
│   └── ping
│       └── ping.pb.go
└── server.go2 directories, 6 files
http://www.yayakq.cn/news/848552/

相关文章：

  • 自建国外购物网站哪些网站做外贸
  • ui设计在线网站楼盘网站建设方案ppt
  • 哈尔滨阿城网站建设wordpress 子目录
  • 网站建设流程时间表影视投资销售怎么找客户
  • 深圳各大网站制作哪家公司好单片机程序员开发网站
  • 协会网站设计方案模板濮阳做公司网站
  • 江苏省建设厅网站福州网站建设推广
  • 少儿编程网站网站模块在线制作
  • 网页设计模板代码网站佛山网站建设推广
  • 怎么做网站主导航网站开发人员没有按照设计开发
  • 公关策划网站建设友情链接网站
  • 美食网站页面设计源代码如何在百度推广网站
  • 公司怎么样做网站360免费建站怎么进不去
  • 网站如何建设名称百度科技有限公司
  • 个人网站如何制作教程wordpress登录破解版
  • 论坛网站前置审批做网站是如何赚钱的
  • 哪个公司建立网站好深圳的小型网络公司
  • 网站的建设时间表周口网站建设专家
  • 做目录网站注意事项如何为wordpress添加音乐播放器
  • 哪个网站可做密丸网站保护等级是企业必须做的么
  • 广东建设工程信息网站成都网站建设价格表
  • 银川网站网站建设中文单页面网站模板
  • 伍佰亿网站专业的vi设计公司
  • php网站换服务器北京网站设计公司飞沐
  • 怎么搭建属于自己的网站做软件去哪个网站
  • 天河区门户网站做直播小视频在线观看网站
  • 游戏网站建设赚钱群辉wordpress
  • 做网站必须购买空间吗合同模板网站
  • 新乡网站建设哪家好外贸黄页
  • 地下城钓鱼网站怎么做一个主机建多少个网站