gRPC的發(fā)布訂閱模式及REST接口和超時控制

更新時間：2022年06月15日 16:51:25 作者：yongxinz

這篇文章主要為大家介紹了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?&amp;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(&amp;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(&amp;proto.String{Value:?v.(string)});?nil?!=?err?{
????????????return?err
????????}
????}
????return?nil
}
func?NewPubsubService()?*PubsubService?{
????return?&amp;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(),?&amp;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(),?&amp;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())
????????}
????}()
????&lt;-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(),?&amp;proto.String{Value:?"golang:?hello?Go"},
????)
????if?err?!=?nil?{
????????log.Fatal(err)
????}
????_,?err?=?client.Publish(
????????context.Background(),?&amp;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?&amp;proto.StringMessage{Value:?"Get?hi:"?+?message.Value?+?"#"},?nil
}
func?(r?*RestServiceImpl)?Post(ctx?context.Context,?message?*proto.StringMessage)?(*proto.StringMessage,?error)?{
????return?&amp;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,?&amp;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?:=?&lt;-data:
????????return?res,?nil
????case?&lt;-ctx.Done():
????????return?nil,?status.Errorf(codes.Canceled,?"Client?cancelled,?abandoning.")
????}
}
func?handle(ctx?context.Context,?req?*proto.HelloRequest,?data?chan&lt;-?*proto.HelloReply)?{
????select?{
????case?&lt;-ctx.Done():
????????log.Println(ctx.Err())
????????runtime.Goexit()?//超時后退出該Go協(xié)程
????case?&lt;-time.After(4?*?time.Second):?//?模擬耗時操作
????????res?:=?proto.HelloReply{
????????????Message:?"hello?"?+?req.Name,
????????}
????????//?//修改數(shù)據(jù)庫前進行超時判斷
????????//?if?ctx.Err()?==?context.Canceled{
????????//??...
????????//??//如果已經超時，則退出
????????//?}
????????data?&lt;-?&amp;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,?&amp;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

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

gRPC的發(fā)布訂閱模式及REST接口和超時控制

目錄

前言

發(fā)布和訂閱模式

proto 文件

服務端

訂閱客戶端

發(fā)布客戶端

REST 接口

proto 文件

REST 服務

gRPC 服務

超時控制

proto 文件

客戶端

服務端

總結

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线 免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

gRPC的發(fā)布訂閱模式及REST接口和超時控制

目錄

前言

發(fā)布和訂閱模式

proto 文件

服務端

訂閱客戶端

發(fā)布客戶端

REST 接口

proto 文件

REST 服務

gRPC 服務

超時控制

proto 文件

客戶端

服務端

總結

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕