gRPC的發(fā)布訂閱模式及REST接口和超時控制
前言
上篇文章 gRPC，爆贊 直接爆了，內容主要包括：簡單的 gRPC 服務，流處理模式，驗證器，Token 認證和證書認證。
在多個平臺的閱讀量都創(chuàng)了新高，在 oschina 更是獲得了首頁推薦，閱讀量到了 1w+，這已經是我單篇閱讀的高峰了。
看來只要用心寫還是有收獲的。
這篇咱們還是從實戰(zhàn)出發(fā)，主要介紹 gRPC 的發(fā)布訂閱模式，REST 接口和超時控制。
相關代碼我會都上傳到 GitHub，感興趣的小伙伴可以去查看或下載。
發(fā)布和訂閱模式
發(fā)布訂閱是一個常見的設計模式，開源社區(qū)中已經存在很多該模式的實現(xiàn)。其中 docker 項目中提供了一個 pubsub 的極簡實現(xiàn)，下面是基于 pubsub 包實現(xiàn)的本地發(fā)布訂閱代碼：
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 創(chuàng)建了一個對象，然后通過 p.SubscribeTopic 實現(xiàn)訂閱，p.Publish 來發(fā)布消息。
執(zhí)行效果如下：
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 包實現(xiàn)發(fā)布訂閱模式。
需要實現(xiàn)四個部分：
- proto 文件；
- 服務端： 用于接收訂閱請求，同時也接收發(fā)布請求，并將發(fā)布請求轉發(fā)給訂閱者；
- 訂閱客戶端： 用于從服務端訂閱消息，處理消息；
- 發(fā)布客戶端： 用于向服務端發(fā)送消息。
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);
}
定義三個方法，分別是一個發(fā)布 Publish 和兩個訂閱 Subscribe 和 SubscribeTopic。
Subscribe 方法接收全部消息，而 SubscribeTopic 根據(jù)特定的 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)
????//?注冊業(yè)務服務
????proto.RegisterPubsubServiceServer(server,?NewPubsubService())
????fmt.Println("grpc?server?start?...")
????if?err?:=?server.Serve(lis);?err?!=?nil?{
????????log.Fatalf("failed?to?serve:?%v",?err)
????}
}
對比之前的發(fā)布訂閱程序，其實這里是將 *pubsub.Publisher 作為了 gRPC 的結構體 PubsubService 的一個成員。
然后還是按照 gRPC 的開發(fā)流程，實現(xiàn)結構體對應的三個方法。
最后，在注冊服務時，將 NewPubsubService() 服務注入，實現(xiàn)本地發(fā)布訂閱功能。
訂閱客戶端
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 對象，然后分別實現(xiàn) client.Subscribe 和 client.SubscribeTopic 方法，再通過 goroutine 不停接收消息。
發(fā)布客戶端
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 方法發(fā)布消息。
當代碼全部寫好之后，我們開三個終端來測試一下：
終端1 上啟動服務端：
go?run?main.go
終端2 上啟動訂閱客戶端：
go?run?sub_client.go
終端3 上執(zhí)行發(fā)布客戶端：
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 文件
第一步還是創(chuàng)建一個 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，分別實現(xiàn) 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 參數(shù)可生成對應的 gw 文件，--swagger_out 參數(shù)可生成對應的 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)
}
這里主要是通過實現(xiàn) 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 服務的實現(xiàn)方式還是和以前一樣。
以上就是全部代碼，現(xiàn)在來測試一下：
啟動三個終端：
終端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"}
源碼地址：
https://github.com/yongxinz/go-example/tree/main/grpc-example/rest
超時控制
最后一部分介紹一下超時控制，這部分內容是非常重要的。
一般的 WEB 服務 API，或者是 Nginx 都會設置一個超時時間，超過這個時間，如果還沒有數(shù)據(jù)返回，服務端可能直接返回一個超時錯誤，或者客戶端也可能結束這個連接。
如果沒有這個超時時間，那是相當危險的。所有請求都阻塞在服務端，會消耗大量資源，比如內存。如果資源耗盡的話，甚至可能會導致整個服務崩潰。
那么，在 gRPC 中怎么設置超時時間呢？主要是通過上下文 context.Context 參數(shù)，具體來說就是 context.WithDeadline 函數(shù)。
proto 文件
創(chuàng)建最簡單的 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)
}
通過下面的函數(shù)設置一個 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協(xié)程
????case?<-time.After(4?*?time.Second):?//?模擬耗時操作
????????res?:=?proto.HelloReply{
????????????Message:?"hello?"?+?req.Name,
????????}
????????//?//修改數(shù)據(jù)庫前進行超時判斷
????????//?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)
????//?注冊業(yè)務服務
????proto.RegisterGreeterServer(server,?&greeter{})
????fmt.Println("grpc?server?start?...")
????if?err?:=?server.Serve(lis);?err?!=?nil?{
????????log.Fatalf("failed?to?serve:?%v",?err)
????}
}
服務端增加一個 handle 函數(shù)，其中 case <-time.After(4 * time.Second) 表示 4s 之后才會執(zhí)行其對應代碼，用來模擬超時請求。
如果客戶端超時時間超過 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
源碼地址
https://github.com/yongxinz/go-example/tree/main/grpc-example/deadline
總結
本文主要介紹了 gRPC 的三部分實戰(zhàn)內容，分別是：
- 發(fā)布訂閱模式
- REST 接口
- 超時控制
個人感覺，超時控制還是最重要的，在平時的開發(fā)過程中需要多多注意。
結合上篇文章，gRPC 的實戰(zhàn)內容就寫完了，代碼全部可以執(zhí)行，也都上傳到了 GitHub。
大家如果有任何疑問，歡迎給我留言，如果感覺不錯的話，也歡迎關注和轉發(fā)。
題圖： 該圖片由 Reytschl 在 Pixabay 上發(fā)布
源碼地址：
https://github.com/yongxinz/go-example
https://github.com/yongxinz/gopher
推薦閱讀
參考鏈接
https://chai2010.cn/advanced-go-programming-book/ch4-rpc/readme.html
http://dhdzp.com/article/210874.htm
更多關于gRPC發(fā)布訂閱REST接口的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要給大家介紹了Golang token的生成和解析,文中通過代碼示例給大家介紹的非常詳細,對大家的學習或工作有一定的幫助,需要的朋友可以參考下2024-02-02
這篇文章主要介紹了golang 實現(xiàn)Location跳轉方式，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-05-05
使用GO語言實現(xiàn)Mysql數(shù)據(jù)庫CURD的簡單示例
本文主要介紹了使用GO語言實現(xiàn)Mysql數(shù)據(jù)庫CURD的簡單示例，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-08-08
這篇文章主要為大家介紹了Go 函數(shù)返回nil遇到的避坑問題分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-01-01
這篇文章主要為大家介紹了Go?Redis客戶端使用對比詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-07-07
這篇文章主要介紹了Go語言的管道Channel用法,實例分析了Go語言中管道的原理與使用技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-02-02
這篇文章主要給大家介紹了關于Golang中time.After的使用理解與釋放問題，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2018-08-08
Go語言保證了既能到達靜態(tài)編譯語言的安全和性能，又達到了動態(tài)語言開發(fā)速度和易維護性，有人形容Go語言：Go=?C?+?Python?,?說明Go語言既有C靜態(tài)語言程序的運行速度，又能達到Python動態(tài)語言的快速開發(fā)，這篇文章主要介紹了Go介紹和環(huán)境配置,需要的朋友可以參考下2022-07-07
目錄遍歷是一個很常見的操作,它的使用場景有如文件目錄查看、文件系統(tǒng)清理、日志分析、項目構建等,本文將詳細介紹在Go中幾種遍歷目錄文件的方法,需要的可以參考下2024-02-02
詳解golang consul-grpc 服務注冊與發(fā)現(xiàn)
這篇文章主要介紹了詳解golang consul-grpc 服務注冊與發(fā)現(xiàn)，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-06-06