gRPC的釋出訂閱模式及REST介面和超時控制

2022-06-15 18:01:55

前言

上篇文章 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 &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)
    // 註冊業務服務
    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(), &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 物件，然後分別實現 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(), &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 方法釋出訊息。

當程式碼全部寫好之後，我們開三個終端來測試一下：

終端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/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 引數可生成對應的 gw 檔案，--swagger_out 引數可生成對應的 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)
}

這裡主要是通過實現 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 服務的實現方式還是和以前一樣。

以上就是全部程式碼，現在來測試一下：

啟動三個終端：

終端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 都會設定一個超時時間，超過這個時間，如果還沒有資料返回，伺服器端可能直接返回一個超時錯誤，或者使用者端也可能結束這個連線。

如果沒有這個超時時間，那是相當危險的。所有請求都阻塞在伺服器端，會消耗大量資源，比如記憶體。如果資源耗盡的話，甚至可能會導致整個服務崩潰。

那麼，在 gRPC 中怎麼設定超時時間呢？主要是通過上下文 context.Context 引數，具體來說就是 context.WithDeadline 函數。

proto 檔案

建立最簡單的 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)
}

通過下面的函數設定一個 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協程
    case &lt;-time.After(4 * time.Second): // 模擬耗時操作
        res := proto.HelloReply{
            Message: "hello " + req.Name,
        }
        // //修改資料庫前進行超時判斷
        // 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)
    // 註冊業務服務
    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 函數，其中 case <-time.After(4 * time.Second) 表示 4s 之後才會執行其對應程式碼，用來模擬超時請求。

如果使用者端超時時間超過 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 的三部分實戰內容，分別是：

釋出訂閱模式
REST 介面
超時控制

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

結合上篇文章，gRPC 的實戰內容就寫完了，程式碼全部可以執行，也都上傳到了 GitHub。

大家如果有任何疑問，歡迎給我留言，如果感覺不錯的話，也歡迎關注和轉發。

題圖：該圖片由 Reytschl 在 Pixabay 上釋出

原始碼地址：

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

https://github.com/yongxinz/gopher

gRPC的釋出訂閱模式及REST介面和超時控制

目錄

前言

釋出和訂閱模式

proto 檔案

伺服器端

訂閱使用者端

釋出使用者端

REST 介面

proto 檔案

REST 服務

gRPC 服務

超時控制

proto 檔案

使用者端

伺服器端

總結

熱門文章