AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

测试小姐姐问我 gRPC 怎么用，我直接把这篇文章甩给了她

2022-04-23 262
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 测试小姐姐问我 gRPC 怎么用，我直接把这篇文章甩给了她

原文链接：测试小姐姐问我 gRPC 怎么用，我直接把这篇文章甩给了她


上篇文章 gRPC，爆赞 直接爆了，内容主要包括：简单的 gRPC 服务，流处理模式，验证器，Token 认证和证书认证。


在多个平台的阅读量都创了新高，在 oschina 更是获得了首页推荐，阅读量到了 1w+，这已经是我单篇阅读的高峰了。


看来只要用心写还是有收获的。


这篇咱们还是从实战出发，主要介绍 gRPC 的发布订阅模式，REST 接口和超时控制。

相关代码我会都上传到 GitHub，感兴趣的小伙伴可以去查看或下载。


发布和订阅模式


发布订阅是一个常见的设计模式，开源社区中已经存在很多该模式的实现。其中 docker 项目中提供了一个 pubsub 的极简实现，下面是基于 pubsub 包实现的本地发布订阅代码：


package main
import (
  "fmt"
  "strings"
  "time"
  "github.com/moby/moby/pkg/pubsub"
)
func main() {
  p := pubsub.NewPublisher(100*time.Millisecond, 10)
  golang := p.SubscribeTopic(func(v interface{}) bool {
    if key, ok := v.(string); ok {
      if strings.HasPrefix(key, "golang:") {
        return true
      }
    }
    return false
  })
  docker := p.SubscribeTopic(func(v interface{}) bool {
    if key, ok := v.(string); ok {
      if strings.HasPrefix(key, "docker:") {
        return true
      }
    }
    return false
  })
  go p.Publish("hi")
  go p.Publish("golang: https://golang.org")
  go p.Publish("docker: https://www.docker.com/")
  time.Sleep(1)
  go func() {
    fmt.Println("golang topic:", <-golang)
  }()
  go func() {
    fmt.Println("docker topic:", <-docker)
  }()
  <-make(chan bool)
}
复制代码


这段代码首先通过 pubsub.NewPublisher 创建了一个对象，然后通过p.SubscribeTopic 实现订阅，p.Publish 来发布消息。


执行效果如下：


docker topic: docker: https://www.docker.com/
golang topic: golang: https://golang.org
fatal error: all goroutines are asleep - deadlock!
goroutine 1 [chan receive]:
main.main()
  /Users/zhangyongxin/src/go-example/grpc-example/pubsub/server/pubsub.go:43 +0x1e7
exit status 2
复制代码


订阅消息可以正常打印。


但有一个死锁报错，是因为这条语句 <-make(chan bool) 引起的。但是如果没有这条语句就不能正常打印订阅消息。


这里就不是很懂了，有没有大佬知道，欢迎留言，求指导。


接下来就用 gRPC 和 pubsub 包实现发布订阅模式。


需要实现四个部分：


  1. proto 文件；
  2. 服务端： 用于接收订阅请求，同时也接收发布请求，并将发布请求转发给订阅者；
  3. 订阅客户端： 用于从服务端订阅消息，处理消息；
  4. 发布客户端： 用于向服务端发送消息。


proto 文件


首先定义 proto 文件：


syntax = "proto3";
package proto;
message String {
    string value = 1;
}
service PubsubService {
    rpc Publish (String) returns (String);
    rpc SubscribeTopic (String) returns (stream String);
    rpc Subscribe (String) returns (stream String);
}
复制代码


定义三个方法，分别是一个发布 Publish 和两个订阅 Subscribe 和 SubscribeTopic。


Subscribe 方法接收全部消息，而 SubscribeTopic 根据特定的 Topic 接收消息。


服务端


package main
import (
  "context"
  "fmt"
  "log"
  "net"
  "server/proto"
  "strings"
  "time"
  "github.com/moby/moby/pkg/pubsub"
  "google.golang.org/grpc"
  "google.golang.org/grpc/reflection"
)
type PubsubService struct {
  pub *pubsub.Publisher
}
func (p *PubsubService) Publish(ctx context.Context, arg *proto.String) (*proto.String, error) {
  p.pub.Publish(arg.GetValue())
  return &proto.String{}, nil
}
func (p *PubsubService) SubscribeTopic(arg *proto.String, stream proto.PubsubService_SubscribeTopicServer) error {
  ch := p.pub.SubscribeTopic(func(v interface{}) bool {
    if key, ok := v.(string); ok {
      if strings.HasPrefix(key, arg.GetValue()) {
        return true
      }
    }
    return false
  })
  for v := range ch {
    if err := stream.Send(&proto.String{Value: v.(string)}); nil != err {
      return err
    }
  }
  return nil
}
func (p *PubsubService) Subscribe(arg *proto.String, stream proto.PubsubService_SubscribeServer) error {
  ch := p.pub.Subscribe()
  for v := range ch {
    if err := stream.Send(&proto.String{Value: v.(string)}); nil != err {
      return err
    }
  }
  return nil
}
func NewPubsubService() *PubsubService {
  return &PubsubService{pub: pubsub.NewPublisher(100*time.Millisecond, 10)}
}
func main() {
  lis, err := net.Listen("tcp", ":50051")
  if err != nil {
    log.Fatalf("failed to listen: %v", err)
  }
  // 简单调用
  server := grpc.NewServer()
  // 注册 grpcurl 所需的 reflection 服务
  reflection.Register(server)
  // 注册业务服务
  proto.RegisterPubsubServiceServer(server, NewPubsubService())
  fmt.Println("grpc server start ...")
  if err := server.Serve(lis); err != nil {
    log.Fatalf("failed to serve: %v", err)
  }
}
复制代码


对比之前的发布订阅程序，其实这里是将 *pubsub.Publisher 作为了 gRPC 的结构体 PubsubService 的一个成员。


然后还是按照 gRPC 的开发流程，实现结构体对应的三个方法。


最后，在注册服务时，将 NewPubsubService() 服务注入，实现本地发布订阅功能。


订阅客户端


package main
import (
  "client/proto"
  "context"
  "fmt"
  "io"
  "log"
  "google.golang.org/grpc"
)
func main() {
  conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
  if err != nil {
    log.Fatal(err)
  }
  defer conn.Close()
  client := proto.NewPubsubServiceClient(conn)
  stream, err := client.Subscribe(
    context.Background(), &proto.String{Value: "golang:"},
  )
  if nil != err {
    log.Fatal(err)
  }
  go func() {
    for {
      reply, err := stream.Recv()
      if nil != err {
        if io.EOF == err {
          break
        }
        log.Fatal(err)
      }
      fmt.Println("sub1: ", reply.GetValue())
    }
  }()
  streamTopic, err := client.SubscribeTopic(
    context.Background(), &proto.String{Value: "golang:"},
  )
  if nil != err {
    log.Fatal(err)
  }
  go func() {
    for {
      reply, err := streamTopic.Recv()
      if nil != err {
        if io.EOF == err {
          break
        }
        log.Fatal(err)
      }
      fmt.Println("subTopic: ", reply.GetValue())
    }
  }()
  <-make(chan bool)
}
复制代码


新建一个 NewPubsubServiceClient 对象，然后分别实现 client.Subscribe 和 client.SubscribeTopic 方法，再通过 goroutine 不停接收消息。


发布客户端


package main
import (
  "client/proto"
  "context"
  "log"
  "google.golang.org/grpc"
)
func main() {
  conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
  if err != nil {
    log.Fatal(err)
  }
  defer conn.Close()
  client := proto.NewPubsubServiceClient(conn)
  _, err = client.Publish(
    context.Background(), &proto.String{Value: "golang: hello Go"},
  )
  if err != nil {
    log.Fatal(err)
  }
  _, err = client.Publish(
    context.Background(), &proto.String{Value: "docker: hello Docker"},
  )
  if nil != err {
    log.Fatal(err)
  }
}
复制代码


新建一个 NewPubsubServiceClient 对象，然后通过 client.Publish 方法发布消息。


当代码全部写好之后，我们开三个终端来测试一下：


终端1 上启动服务端：


go run main.go
复制代码


终端2 上启动订阅客户端：


go run sub_client.go
复制代码


终端3 上执行发布客户端：


go run pub_client.go
复制代码


这样，在 终端2 上就有对应的输出了：


subTopic:  golang: hello Go
sub1:  golang: hello Go
sub1:  docker: hello Docker
复制代码


也可以再多开几个订阅终端，那么每一个订阅终端上都会有相同的内容输出。


源码地址：GitHub


REST 接口


gRPC 一般用于集群内部通信，如果需要对外提供服务，大部分都是通过 REST 接口的方式。开源项目 grpc-gateway 提供了将 gRPC 服务转换成 REST 服务的能力，通过这种方式，就可以直接访问 gRPC API 了。


但我觉得，实际上这么用的应该还是比较少的。如果提供 REST 接口的话，直接写一个 HTTP 服务会方便很多。


proto 文件


第一步还是创建一个 proto 文件：


syntax = "proto3";
package proto;
import "google/api/annotations.proto";
message StringMessage {
  string value = 1;
}
service RestService {
    rpc Get(StringMessage) returns (StringMessage) {
        option (google.api.http) = {
            get: "/get/{value}"
        };
    }
    rpc Post(StringMessage) returns (StringMessage) {
        option (google.api.http) = {
            post: "/post"
            body: "*"
        };
    }
}
复制代码


定义一个 REST 服务 RestService，分别实现 GET 和 POST 方法。


安装插件：


go get -u github.com/grpc-ecosystem/grpc-gateway/protoc-gen-grpc-gateway
复制代码


生成对应代码：


protoc -I/usr/local/include -I. \
    -I$GOPATH/pkg/mod \
    -I$GOPATH/pkg/mod/github.com/grpc-ecosystem/grpc-gateway@v1.16.0/third_party/googleapis \
    --grpc-gateway_out=. --go_out=plugins=grpc:.\
    --swagger_out=. \
    helloworld.proto
复制代码


--grpc-gateway_out 参数可生成对应的 gw 文件，--swagger_out 参数可生成对应的 API 文档。


在我这里生成的两个文件如下：


helloworld.pb.gw.go
helloworld.swagger.json
复制代码


REST 服务


package main
import (
  "context"
  "log"
  "net/http"
  "rest/proto"
  "github.com/grpc-ecosystem/grpc-gateway/runtime"
  "google.golang.org/grpc"
)
func main() {
  ctx := context.Background()
  ctx, cancel := context.WithCancel(ctx)
  defer cancel()
  mux := runtime.NewServeMux()
  err := proto.RegisterRestServiceHandlerFromEndpoint(
    ctx, mux, "localhost:50051",
    []grpc.DialOption{grpc.WithInsecure()},
  )
  if err != nil {
    log.Fatal(err)
  }
  http.ListenAndServe(":8080", mux)
}
复制代码


这里主要是通过实现 gw 文件中的 RegisterRestServiceHandlerFromEndpoint 方法来连接 gRPC 服务。


gRPC 服务


package main
import (
  "context"
  "net"
  "rest/proto"
  "google.golang.org/grpc"
)
type RestServiceImpl struct{}
func (r *RestServiceImpl) Get(ctx context.Context, message *proto.StringMessage) (*proto.StringMessage, error) {
  return &proto.StringMessage{Value: "Get hi:" + message.Value + "#"}, nil
}
func (r *RestServiceImpl) Post(ctx context.Context, message *proto.StringMessage) (*proto.StringMessage, error) {
  return &proto.StringMessage{Value: "Post hi:" + message.Value + "@"}, nil
}
func main() {
  grpcServer := grpc.NewServer()
  proto.RegisterRestServiceServer(grpcServer, new(RestServiceImpl))
  lis, _ := net.Listen("tcp", ":50051")
  grpcServer.Serve(lis)
}
复制代码


gRPC 服务的实现方式还是和以前一样。


以上就是全部代码，现在来测试一下：


启动三个终端：


终端1 启动 gRPC 服务：


go run grpc_service.go
复制代码


终端2 启动 REST 服务：


go run rest_service.go
复制代码


终端3 来请求 REST 服务：


$ curl localhost:8080/get/gopher
{"value":"Get hi:gopher"}
$ curl localhost:8080/post -X POST --data '{"value":"grpc"}'
{"value":"Post hi:grpc"}
复制代码


源码地址：GitHub


超时控制


最后一部分介绍一下超时控制，这部分内容是非常重要的。


一般的 WEB 服务 API，或者是 Nginx 都会设置一个超时时间，超过这个时间，如果还没有数据返回，服务端可能直接返回一个超时错误，或者客户端也可能结束这个连接。


如果没有这个超时时间，那是相当危险的。所有请求都阻塞在服务端，会消耗大量资源，比如内存。如果资源耗尽的话，甚至可能会导致整个服务崩溃。


那么，在 gRPC 中怎么设置超时时间呢？主要是通过上下文 context.Context 参数，具体来说就是 context.WithDeadline 函数。


proto 文件


创建最简单的 proto 文件，这个不多说。


syntax = "proto3";
package proto;
// The greeting service definition.
service Greeter {
    // Sends a greeting
    rpc SayHello (HelloRequest) returns (HelloReply) {}
}
// The request message containing the user's name.
message HelloRequest {
    string name = 1;
}
// The response message containing the greetings
message HelloReply {
    string message = 1;
}
复制代码


客户端


package main
import (
  "client/proto"
  "context"
  "fmt"
  "log"
  "time"
  "google.golang.org/grpc"
  "google.golang.org/grpc/codes"
  "google.golang.org/grpc/status"
)
func main() {
  // 简单调用
  conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
  defer conn.Close()
  ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Duration(3*time.Second)))
  defer cancel()
  client := proto.NewGreeterClient(conn)
  // 简单调用
  reply, err := client.SayHello(ctx, &proto.HelloRequest{Name: "zzz"})
  if err != nil {
    statusErr, ok := status.FromError(err)
    if ok {
      if statusErr.Code() == codes.DeadlineExceeded {
        log.Fatalln("client.SayHello err: deadline")
      }
    }
    log.Fatalf("client.SayHello err: %v", err)
  }
  fmt.Println(reply.Message)
}
复制代码


通过下面的函数设置一个 3s 的超时时间：


ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Duration(3*time.Second)))
defer cancel()
复制代码


然后在响应错误中对超时错误进行检测。


服务端


package main
import (
  "context"
  "fmt"
  "log"
  "net"
  "runtime"
  "server/proto"
  "time"
  "google.golang.org/grpc"
  "google.golang.org/grpc/codes"
  "google.golang.org/grpc/reflection"
  "google.golang.org/grpc/status"
)
type greeter struct {
}
func (*greeter) SayHello(ctx context.Context, req *proto.HelloRequest) (*proto.HelloReply, error) {
  data := make(chan *proto.HelloReply, 1)
  go handle(ctx, req, data)
  select {
  case res := <-data:
    return res, nil
  case <-ctx.Done():
    return nil, status.Errorf(codes.Canceled, "Client cancelled, abandoning.")
  }
}
func handle(ctx context.Context, req *proto.HelloRequest, data chan<- *proto.HelloReply) {
  select {
  case <-ctx.Done():
    log.Println(ctx.Err())
    runtime.Goexit() //超时后退出该Go协程
  case <-time.After(4 * time.Second): // 模拟耗时操作
    res := proto.HelloReply{
      Message: "hello " + req.Name,
    }
    // //修改数据库前进行超时判断
    // if ctx.Err() == context.Canceled{
    //  ...
    //  //如果已经超时，则退出
    // }
    data <- &res
  }
}
func main() {
  lis, err := net.Listen("tcp", ":50051")
  if err != nil {
    log.Fatalf("failed to listen: %v", err)
  }
  // 简单调用
  server := grpc.NewServer()
  // 注册 grpcurl 所需的 reflection 服务
  reflection.Register(server)
  // 注册业务服务
  proto.RegisterGreeterServer(server, &greeter{})
  fmt.Println("grpc server start ...")
  if err := server.Serve(lis); err != nil {
    log.Fatalf("failed to serve: %v", err)
  }
}
复制代码


服务端增加一个 handle 函数，其中 case <-time.After(4 * time.Second) 表示 4s 之后才会执行其对应代码，用来模拟超时请求。


如果客户端超时时间超过 4s 的话，就会产生超时报错。


下面来模拟一下：


服务端：


$ go run main.go
grpc server start ...
2021/10/24 22:57:40 context deadline exceeded
复制代码


客户端：


$ go run main.go
2021/10/24 22:57:40 client.SayHello err: deadline
exit status 1
复制代码


源码地址：GitHub


总结


本文主要介绍了 gRPC 的三部分实战内容，分别是：


  1. 发布订阅模式
  2. REST 接口
  3. 超时控制


个人感觉，超时控制还是最重要的，在平时的开发过程中需要多多注意。


结合上篇文章，gRPC 的实战内容就写完了，代码全部可以执行，也都上传到了

GitHub。



源码地址：



文章标签：
Go
容器
应用服务中间件
网络架构
Docker
nginx
API
设计模式
关键词：
测试grpc
alwaysbeta
目录
相关文章
@北鲲
|
5月前
|
SQL 安全 关系型数据库
接上篇文章，在测试宝塔 WAF 的未授权访问漏洞时无意间还发现了一个 SQL 注入漏洞
接上篇文章，在测试宝塔 WAF 的未授权访问漏洞时无意间还发现了一个 SQL 注入漏洞，品相还不错，可执行任意 SQL 语句。 总之，吃了一惊，一个防 SQL 注入的工具居然也有 SQL 注入漏洞。 请看这段代码
@北鲲
561 5
IT测试之美
|
测试技术
读完这篇文章后，才发现Allpairs这款工具，让测试用例变得如此简单
读完这篇文章后，才发现Allpairs这款工具，让测试用例变得如此简单
IT测试之美
394 0
TesterMuller
|
SQL 数据采集 关系型数据库
软件测试|一篇文章带你深入理解SQL约束
软件测试|一篇文章带你深入理解SQL约束
TesterMuller
90 0
众所周知
|
3月前
|
测试技术 API 持续交付
【Python自动化测试】文章探讨了Python在测试领域的关键作用，分为三部分
【7月更文挑战第2天】【Python自动化测试】文章探讨了Python在测试领域的关键作用，分为三部分：1) 自动化测试的重要性与Python的易用性、库支持、跨平台和社区优势；2) Unittest作为标准测试框架的基础用法，及Pytest的灵活性与强大功能；3) 实践中包括Selenium的Web UI测试、Requests的API测试，强调测试隔离、持续集成等最佳实践。Python助力高效稳定的软件测试。
众所周知
50 2
n7lskg6uuowc2
|
4月前
|
监控 Shell 测试技术
一篇文章讲明白MonkeyAPP压力稳定性测试
一篇文章讲明白MonkeyAPP压力稳定性测试
n7lskg6uuowc2
179 1
游客vnj3lgtpj3l56
|
3月前
|
机器学习/深度学习 存储 Linux
我的OS Copilot测试文章
在阿里云管理控制台尝试重置密码时遇到错误，未能成功跳转至相应实例；但后续的密码重置流程顺利，安全组设置和SSH连接快速无阻，且体验到AccessKey预警系统的实用性。在处理代码时曾因未替换特殊字符报错，但最终解决了问题，实现了Python代码分析功能，确认系统为Linux，并对OS工具的交互性和实用性表示赞赏，特别是其与Linux环境的兼容性。遇到过一次生成实例的错误，但整体体验良好，有兴趣参与OS Copilot的开源开发。
游客vnj3lgtpj3l56
76 0
wishCoding
|
4月前
|
JavaScript Java 测试技术
基于ssm+vue.js+uniapp小程序的高中信息技术课程在线测试系统附带文章和源代码部署视频讲解等
基于ssm+vue.js+uniapp小程序的高中信息技术课程在线测试系统附带文章和源代码部署视频讲解等
wishCoding
41 6
TesterMuller
|
3月前
|
运维 测试技术 双11
什么是性能测试，一篇文章告诉你！
性能测试评估系统在现实负载下的性能和可靠性，包括响应时间、吞吐量和稳定性。目的是发现瓶颈、评估系统能力、优化性能和确保可靠性。在**双十一大促**等高并发场景下，性能测试至关重要。它有助于合理规划资源，降低成本，提升效率。测试工程师需掌握性能调优，理解压力曲线图，识别最佳并发用户数和最大承载点。通过测试，确保系统在最佳效率下运行，避免资源浪费和用户满意度下降。
TesterMuller
55 0
67vpym2vvkfhs
|
4月前
|
Web App开发 测试技术 iOS开发
一篇文章讲明白locust性能测试
一篇文章讲明白locust性能测试
67vpym2vvkfhs
54 0
wishCoding
|
5月前
|
JavaScript Java 测试技术
基于ssm+vue.js的在线测试管理系统附带文章和源代码设计说明文档ppt
基于ssm+vue.js的在线测试管理系统附带文章和源代码设计说明文档ppt
wishCoding
40 0

热门文章

最新文章

  • 1
    麒麟开源堡垒机安装部署测试及优缺点总结
  • 2
    深度 ｜ 大数据算法应用的测试发展之路
  • 3
    【精准测试】iOS 代码覆盖率数据采集自动化实践
  • 4
    Oracle Database 12c(12.1) Beta已经开始内部测试
  • 5
    phpunit测试学习 2 分类总结断言涉及哪些方面
  • 6
    使用Android、S3C6410开发板和Ubuntu测试Linux驱动
  • 7
    利用两种不同的NAT配置实现两个接口的双向静态NAT测试
  • 8
    开放源代码与.NET应用程序平台的性能测试
  • 9
    高通发布实地5G网络测试结果，8K视频观看毫无压力
  • 10
    亚马逊测试人工智能云服务 与谷歌微软等竞争
  • 1
    这10款性能测试工具，收藏起来，测试人的工具箱！
    318
  • 2
    如何实现对iframe的自动化测试，一篇文章告诉你
    47
  • 3
    视觉智能平台常见问题之体验产品的美颜测试关掉如何解决
    57
  • 4
    Apache Flink官方网站提供了关于如何使用Docker进行Flink CDC测试的文档
    648
  • 5
    Kafka【付诸实践 03】Offset Explorer Kafka 的终极 UI 工具安装+简单上手+关键特性测试（一篇学会使用 Offset Explorer）
    724
  • 6
    P2P网络下分布式文件共享场景的测试
    273
  • 7
    Jest测试核心
    37
  • 8
    探索自动化测试在持续集成环境中的关键角色
    51
  • 9
    软件质量测试笔记-合工大
    154
  • 10
    模拟试错（STE）法让7B大模型测试超GPT-4
    116

    • 相关课程

    更多
  • MSE微服务测试最佳实践 - 自动化回归
  • 机器阅读技术与应用
  • PHP进阶教程 - 由浅入深掌握面向对象开发 - 第二阶段
  • Java面试疑难点解析 - 系统架构及项目设计
  • SQL基础开发与应用
  • Go语言完全自学手册图文教程

    • 相关电子书

    更多
  • 移动互联网测试到质量的转变
  • 给ITer的技术实战进阶课-阿里CIO学院独家教材（四）
  • F2etest — 多浏览器兼容性测试整体解决方案

    • 相关实验场景

    更多
  • 基于小程序Serverless开发个人相册小程序
  • 1分钟部署经典小游戏
  • 快速上手并跑通AnalyticDB PostgreSQL版TPC-H测试
  • 基于函数计算快速搭建Zblog等传统应用框架
  • 灵活弹性的PolarDB Serverless确保数据业务持续在线
  • Arthas 进阶教程
    • 下一篇
    无影云桌面