測(cè)試小姐姐問(wèn)我 gRPC 怎么用，我直接把這篇文章甩給了她

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

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

看來(lái)只要用心寫(xiě)還是有收獲的。

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

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

發(fā)布和訂閱模式

發(fā)布訂閱是一個(gè)常見(jiàn)的設(shè)計(jì)模式，開(kāi)源社區(qū)中已經(jīng)存在很多該模式的實(shí)現(xiàn)。其中 docker 項(xiàng)目中提供了一個(gè) pubsub 的極簡(jiǎn)實(shí)現(xiàn)，下面是基于 pubsub 包實(shí)現(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)
}

這段代碼首先通過(guò) pubsub.NewPublisher 創(chuàng)建了一個(gè)對(duì)象，然后通過(guò) p.SubscribeTopic 實(shí)現(xiàn)訂閱，p.Publish 來(lái)發(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

訂閱消息可以正常打印。

但有一個(gè)死鎖報(bào)錯(cuò)，是因?yàn)檫@條語(yǔ)句 <-make(chan bool) 引起的。但是如果沒(méi)有這條語(yǔ)句就不能正常打印訂閱消息。

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

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

需要實(shí)現(xiàn)四個(gè)部分：

1. proto 文件；

2. 服務(wù)端： 用于接收訂閱請(qǐng)求，同時(shí)也接收發(fā)布請(qǐng)求，并將發(fā)布請(qǐng)求轉(zhuǎn)發(fā)給訂閱者；

3. 訂閱客戶(hù)端： 用于從服務(wù)端訂閱消息，處理消息；

4. 發(fā)布客戶(hù)端： 用于向服務(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);
}

定義三個(gè)方法，分別是一個(gè)發(fā)布 Publish 和兩個(gè)訂閱 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?&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)
????}

????//?簡(jiǎn)單調(diào)用
????server?:=?grpc.NewServer()
????//?注冊(cè)?grpcurl?所需的?reflection?服務(wù)
????reflection.Register(server)
????//?注冊(cè)業(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)
????}
}

對(duì)比之前的發(fā)布訂閱程序，其實(shí)這里是將 *pubsub.Publisher 作為了 gRPC 的結(jié)構(gòu)體 PubsubService 的一個(gè)成員。

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

最后，在注冊(cè)服務(wù)時(shí)，將 NewPubsubService() 服務(wù)注入，實(shí)現(xiàn)本地發(fā)布訂閱功能。

訂閱客戶(hù)端

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)
}

新建一個(gè) NewPubsubServiceClient 對(duì)象，然后分別實(shí)現(xiàn) client.Subscribe 和 client.SubscribeTopic 方法，再通過(guò) goroutine 不停接收消息。

發(fā)布客戶(hù)端

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)
????}

}

新建一個(gè) NewPubsubServiceClient 對(duì)象，然后通過(guò) client.Publish 方法發(fā)布消息。

當(dāng)代碼全部寫(xiě)好之后，我們開(kāi)三個(gè)終端來(lái)測(cè)試一下：

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

go?run?main.go

終端2 上啟動(dòng)訂閱客戶(hù)端：

go?run?sub_client.go

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

go?run?pub_client.go

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

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

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

源碼地址：?https://github.com/yongxinz/go-example/tree/main/grpc-example/pubsub

REST 接口

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

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

proto 文件

第一步還是創(chuàng)建一個(gè) 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:?"*"
????????};
????}
}

定義一個(gè) REST 服務(wù) RestService，分別實(shí)現(xiàn) GET 和 POST 方法。

安裝插件：

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

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

在我這里生成的兩個(gè)文件如下：

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)
}

這里主要是通過(guò)實(shí)現(xiàn) gw 文件中的 RegisterRestServiceHandlerFromEndpoint 方法來(lái)連接 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?&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 服務(wù)的實(shí)現(xiàn)方式還是和以前一樣。

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

啟動(dòng)三個(gè)終端：

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

go?run?grpc_service.go

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

go?run?rest_service.go

終端3 來(lái)請(qǐng)求 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

超時(shí)控制

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

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

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

那么，在 gRPC 中怎么設(shè)置超時(shí)時(shí)間呢？主要是通過(guò)上下文 context.Context 參數(shù)，具體來(lái)說(shuō)就是 context.WithDeadline 函數(shù)。

proto 文件

創(chuàng)建最簡(jiǎn)單的 proto 文件，這個(gè)不多說(shuō)。

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;
}

客戶(hù)端

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()?{
????//?簡(jiǎn)單調(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)
????//?簡(jiǎn)單調(diào)用
????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)
}

通過(guò)下面的函數(shù)設(shè)置一個(gè) 3s 的超時(shí)時(shí)間：

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

然后在響應(yīng)錯(cuò)誤中對(duì)超時(shí)錯(cuò)誤進(jìn)行檢測(cè)。

服務(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?:=?<-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()?//超時(shí)后退出該Go協(xié)程
????case?<-time.After(4?*?time.Second):?//?模擬耗時(shí)操作
????????res?:=?proto.HelloReply{
????????????Message:?"hello?"?+?req.Name,
????????}
????????//?//修改數(shù)據(jù)庫(kù)前進(jìn)行超時(shí)判斷
????????//?if?ctx.Err()?==?context.Canceled{
????????//??...
????????//??//如果已經(jīng)超時(shí)，則退出
????????//?}
????????data?<-?&res
????}
}

func?main()?{
????lis,?err?:=?net.Listen("tcp",?":50051")
????if?err?!=?nil?{
????????log.Fatalf("failed?to?listen:?%v",?err)
????}

????//?簡(jiǎn)單調(diào)用
????server?:=?grpc.NewServer()
????//?注冊(cè)?grpcurl?所需的?reflection?服務(wù)
????reflection.Register(server)
????//?注冊(cè)業(yè)務(wù)服務(wù)
????proto.RegisterGreeterServer(server,?&greeter{})

????fmt.Println("grpc?server?start?...")
????if?err?:=?server.Serve(lis);?err?!=?nil?{
????????log.Fatalf("failed?to?serve:?%v",?err)
????}
}

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

如果客戶(hù)端超時(shí)時(shí)間超過(guò) 4s 的話(huà)，就會(huì)產(chǎn)生超時(shí)報(bào)錯(cuò)。

下面來(lái)模擬一下：

服務(wù)端：

$?go?run?main.go
grpc?server?start?...
2021/10/24?22:57:40?context?deadline?exceeded

客戶(hù)端：

$?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 的三部分實(shí)戰(zhàn)內(nèi)容，分別是：

1. 發(fā)布訂閱模式

2. REST 接口

3. 超時(shí)控制

個(gè)人感覺(jué)，超時(shí)控制還是最重要的，在平時(shí)的開(kāi)發(fā)過(guò)程中需要多多注意。

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

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

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

源碼地址：

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

• https://github.com/yongxinz/gopher

關(guān)注公眾號(hào) AlwaysBeta，回復(fù)「goebook」領(lǐng)取 Go 編程經(jīng)典書(shū)籍。

推薦閱讀：

• gRPC，爆贊

• 使用 grpcurl 通過(guò)命令行訪(fǎng)問(wèn) gRPC 服務(wù)

• 聽(tīng)說(shuō)，99% 的 Go 程序員都被 defer 坑過(guò)

參考：

• https://chai2010.cn/advanced-go-programming-book/ch4-rpc/readme.html

• https://codeleading.com/article/94674952433/

• https://juejin.cn/post/6844904017017962504

• https://www.cnblogs.com/FireworksEasyCool/p/12702959.html