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

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

這篇文章主要為大家介紹了gRPC的發(fā)布訂閱模式及REST接口和超時控制，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

前言

上篇文章 gRPC，爆贊直接爆了，內(nèi)容主要包括：簡單的 gRPC 服務(wù)，流處理模式，驗證器，Token 認(rèn)證和證書認(rèn)證。

在多個平臺的閱讀量都創(chuàng)了新高，在 oschina 更是獲得了首頁推薦，閱讀量到了 1w+，這已經(jīng)是我單篇閱讀的高峰了。

看來只要用心寫還是有收獲的。

這篇咱們還是從實戰(zhàn)出發(fā)，主要介紹 gRPC 的發(fā)布訂閱模式，REST 接口和超時控制。

相關(guān)代碼我會都上傳到 GitHub，感興趣的小伙伴可以去查看或下載。

發(fā)布和訂閱模式

發(fā)布訂閱是一個常見的設(shè)計模式，開源社區(qū)中已經(jīng)存在很多該模式的實現(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) 引起的。但是如果沒有這條語句就不能正常打印訂閱消息。

這里就不是很懂了，有沒有大佬知道，歡迎留言，求指導(dǎo)。

接下來就用 gRPC 和 pubsub 包實現(xiàn)發(fā)布訂閱模式。

需要實現(xiàn)四個部分：

proto 文件；
服務(wù)端：用于接收訂閱請求，同時也接收發(fā)布請求，并將發(fā)布請求轉(zhuǎn)發(fā)給訂閱者；
訂閱客戶端：用于從服務(wù)端訂閱消息，處理消息；
發(fā)布客戶端：用于向服務(wù)端發(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 接收消息。

服務(wù)端

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)
????}
????//?簡單調(diào)用
????server?:=?grpc.NewServer()
????//?注冊?grpcurl?所需的?reflection?服務(wù)
????reflection.Register(server)
????//?注冊業(yè)務(wù)服務(wù)
????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 的結(jié)構(gòu)體 PubsubService 的一個成員。

然后還是按照 gRPC 的開發(fā)流程，實現(xiàn)結(jié)構(gòu)體對應(yīng)的三個方法。

最后，在注冊服務(wù)時，將 NewPubsubService() 服務(wù)注入，實現(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ā)布消息。

當(dāng)代碼全部寫好之后，我們開三個終端來測試一下：

終端1 上啟動服務(wù)端：

go?run?main.go

終端2 上啟動訂閱客戶端：

go?run?sub_client.go

終端3 上執(zhí)行發(fā)布客戶端：

go?run?pub_client.go

這樣，在 終端2 上就有對應(yīng)的輸出了：

subTopic:  golang: hello Go
sub1:  golang: hello Go
sub1:  docker: hello Docker

也可以再多開幾個訂閱終端，那么每一個訂閱終端上都會有相同的內(nèi)容輸出。

源碼地址：

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

REST 接口

gRPC 一般用于集群內(nèi)部通信，如果需要對外提供服務(wù)，大部分都是通過 REST 接口的方式。開源項目 grpc-gateway 提供了將 gRPC 服務(wù)轉(zhuǎn)換成 REST 服務(wù)的能力，通過這種方式，就可以直接訪問 gRPC API 了。

但我覺得，實際上這么用的應(yīng)該還是比較少的。如果提供 REST 接口的話，直接寫一個 HTTP 服務(wù)會方便很多。

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 服務(wù) RestService，分別實現(xiàn) GET 和 POST 方法。

安裝插件：

go?get?-u?github.com/grpc-ecosystem/grpc-gateway/protoc-gen-grpc-gateway

生成對應(yīng)代碼：

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ù)可生成對應(yīng)的 gw 文件，--swagger_out 參數(shù)可生成對應(yīng)的 API 文檔。

在我這里生成的兩個文件如下：

helloworld.pb.gw.go
helloworld.swagger.json

REST 服務(wù)

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 服務(wù)。

gRPC 服務(wù)

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 服務(wù)的實現(xiàn)方式還是和以前一樣。

以上就是全部代碼，現(xiàn)在來測試一下：

啟動三個終端：

終端1 啟動 gRPC 服務(wù)：

go?run?grpc_service.go

終端2 啟動 REST 服務(wù)：

go?run?rest_service.go

終端3 來請求 REST 服務(wù)：

$?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

超時控制

最后一部分介紹一下超時控制，這部分內(nèi)容是非常重要的。

一般的 WEB 服務(wù) API，或者是 Nginx 都會設(shè)置一個超時時間，超過這個時間，如果還沒有數(shù)據(jù)返回，服務(wù)端可能直接返回一個超時錯誤，或者客戶端也可能結(jié)束這個連接。

如果沒有這個超時時間，那是相當(dāng)危險的。所有請求都阻塞在服務(wù)端，會消耗大量資源，比如內(nèi)存。如果資源耗盡的話，甚至可能會導(dǎo)致整個服務(wù)崩潰。

那么，在 gRPC 中怎么設(shè)置超時時間呢？主要是通過上下文 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()?{
????//?簡單調(diào)用
????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)
????//?簡單調(diào)用
????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ù)設(shè)置一個 3s 的超時時間：

ctx,?cancel?:=?context.WithDeadline(context.Background(),?time.Now().Add(time.Duration(3*time.Second)))
defer?cancel()

然后在響應(yīng)錯誤中對超時錯誤進行檢測。

服務(wù)端

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{
????????//??...
????????//??//如果已經(jīng)超時，則退出
????????//?}
????????data?&lt;-?&amp;res
????}
}
func?main()?{
????lis,?err?:=?net.Listen("tcp",?":50051")
????if?err?!=?nil?{
????????log.Fatalf("failed?to?listen:?%v",?err)
????}
????//?簡單調(diào)用
????server?:=?grpc.NewServer()
????//?注冊?grpcurl?所需的?reflection?服務(wù)
????reflection.Register(server)
????//?注冊業(yè)務(wù)服務(wù)
????proto.RegisterGreeterServer(server,?&amp;greeter{})
????fmt.Println("grpc?server?start?...")
????if?err?:=?server.Serve(lis);?err?!=?nil?{
????????log.Fatalf("failed?to?serve:?%v",?err)
????}
}

服務(wù)端增加一個 handle 函數(shù)，其中 case <-time.After(4 * time.Second) 表示 4s 之后才會執(zhí)行其對應(yīng)代碼，用來模擬超時請求。

如果客戶端超時時間超過 4s 的話，就會產(chǎn)生超時報錯。

下面來模擬一下：

服務(wù)端：

$?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

總結(jié)

本文主要介紹了 gRPC 的三部分實戰(zhàn)內(nèi)容，分別是：

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

個人感覺，超時控制還是最重要的，在平時的開發(fā)過程中需要多多注意。

結(jié)合上篇文章，gRPC 的實戰(zhàn)內(nèi)容就寫完了，代碼全部可以執(zhí)行，也都上傳到了 GitHub。

大家如果有任何疑問，歡迎給我留言，如果感覺不錯的話，也歡迎關(guān)注和轉(zhuǎn)發(fā)。

題圖：該圖片由 Reytschl 在 Pixabay 上發(fā)布

源碼地址：

https://github.com/yongxinz/go-example

https://github.com/yongxinz/gopher

欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

軟件下載

源碼下載

軟件編程

網(wǎng)絡(luò)編程

在線工具

數(shù)據(jù)庫

CMS

常用工具

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

目錄

前言

發(fā)布和訂閱模式

proto 文件

服務(wù)端

訂閱客戶端

發(fā)布客戶端

REST 接口

proto 文件

REST 服務(wù)

gRPC 服務(wù)

超時控制

proto 文件

客戶端

服務(wù)端

總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具