您现在的位置是：网站首页> 编程资料编程资料

gRPC的发布订阅模式及REST接口和超时控制_Golang_

2023-05-26 541人已围观

简介 gRPC的发布订阅模式及REST接口和超时控制_Golang_

前言

上篇文章 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 包实现发布订阅模式。

需要实现四个部分：

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

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

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

源码地址：

https://github.com/yongxinz/go-example/tree/main/grpc-example/pubsub

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/m
                
                

                

                                            
上一篇：Golang实现程序优雅退出的方法详解_Golang_

                                                                
下一篇：gRPC超时拦截器实现示例_Golang_

相关内容

点击排行

本栏推荐

猜你喜欢

-六神源码网