[[430935]]

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

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

發(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. 訂閱客戶端： 用于從服務(wù)端訂閱消息，處理消息;
4. 發(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); 
} 

定義三個(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ā)布訂閱功能。

訂閱客戶端

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ā)布客戶端

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)訂閱客戶端：

go run sub_client.go 

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

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ò)這種方式，就可以直接訪問(wèn) gRPC API 了。

但我覺(jué)得，實(shí)際上這么用的應(yīng)該還是比較少的。如果提供 REST 接口的話，直接寫(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ò)誤，或者客戶端也可能結(jié)束這個(gè)連接。

如果沒(méi)有這個(gè)超時(shí)時(shí)間，那是相當(dāng)危險(xiǎn)的。所有請(qǐng)求都阻塞在服務(wù)端，會(huì)消耗大量資源，比如內(nèi)存。如果資源耗盡的話，甚至可能會(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; 
} 

客戶端

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)求。

如果客戶端超時(shí)時(shí)間超過(guò) 4s 的話，就會(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 

客戶端：

$ 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ò)程中需要多多注意。

本文轉(zhuǎn)載自微信公眾號(hào)「AlwaysBeta」