Go 语言实现 gRPC 的发布订阅模式，REST 接口和超时控制

2023年5月25日下午10:33 • Go语言 • 阅读 71

在多个平台的阅读量都创了新高，在 oschina 更是获得了首页推荐，阅读量到了 1w+，这已经是我单篇阅读的高峰了。

看来只要用心写还是有收获的。

这篇咱们还是从实战出发，主要介绍 gRPC 的发布订阅模式，REST 接口和超时控制。

发布订阅是一个常见的设计模式，开源社区中已经存在很多该模式的实现。其中 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:",

执行效果如下：

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)< code> 引起的。但是如果没有这条语句就不能正常打印订阅消息。

这里就不是很懂了，有没有大佬知道，欢迎留言，求指导。

接下来就用 gRPC 和 pubsub 包实现发布订阅模式。

需要实现四个部分：

首先定义 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 &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)
    }

    // 简单调用
    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(), &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())
        }
    }()

新建一个 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(), &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)
    }

}

新建一个 NewPubsubServiceClient 对象，然后通过 client.Publish 方法发布消息。

当代码全部写好之后，我们开三个终端来测试一下：

终端1 上启动服务端：

go run main.go

终端2 上启动订阅客户端：

go run sub_client.go

终端3 上执行发布客户端：

go run pub_client.go

这样，在 终端2 上就有对应的输出了：

也可以再多开几个订阅终端，那么每一个订阅终端上都会有相同的内容输出。

gRPC 一般用于集群内部通信，如果需要对外提供服务，大部分都是通过 REST 接口的方式。开源项目 grpc-gateway 提供了将 gRPC 服务转换成 REST 服务的能力，通过这种方式，就可以直接访问 gRPC API 了。

但我觉得，实际上这么用的应该还是比较少的。如果提供 REST 接口的话，直接写一个 HTTP 服务会方便很多。

第一步还是创建一个 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

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 服务。

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 服务的实现方式还是和以前一样。

以上就是全部代码，现在来测试一下：

启动三个终端：

终端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"}

最后一部分介绍一下超时控制，这部分内容是非常重要的。

一般的 WEB 服务 API，或者是 Nginx 都会设置一个超时时间，超过这个时间，如果还没有数据返回，服务端可能直接返回一个超时错误，或者客户端也可能结束这个连接。

如果没有这个超时时间，那是相当危险的。所有请求都阻塞在服务端，会消耗大量资源，比如内存。如果资源耗尽的话，甚至可能会导致整个服务崩溃。

创建最简单的 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, &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 :=

服务端增加一个 handle 函数，其中 case <-time.after(4 * time.second)< code> 表示 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

本文主要介绍了 gRPC 的三部分实战内容，分别是：

个人感觉，超时控制还是最重要的，在平时的开发过程中需要多多注意。

大家如果有任何疑问，欢迎给我留言，如果感觉不错的话，也欢迎关注和转发。

源码地址：

推荐阅读：

参考：

Original: https://www.cnblogs.com/alwaysbeta/p/15469057.html
Author: yongxinz
Title: Go 语言实现 gRPC 的发布订阅模式，REST 接口和超时控制

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/516546/

转载文章受原作者版权保护。转载请注明原作者出处！

Go语言

【自取】最近整理的，有需要可以领取学习：

Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言刷题答案下载！

新作：轻量级Golang IoC容器——iocgo

习惯于Java或者C#开发的人应该对控制反转与依赖注入应该再熟悉不过了。在Java平台有鼎鼎大名的Spring框架，在C#平台有Autofac,Unity,Windsor等，我当年…

Go语言 2023年5月25日
0080
第十五章：指针类型

本篇翻译自《Practical Go Lessons》 Chapter 15: Pointer type 1 你将在本章将学到什么？什么是指针？什么时指针类型？如何去创建并使…

Go语言 2023年5月25日
0067
编译kubeadm使生成证书有效期为100年

问题编译检查结果问题当我使用kubeadm部署成功k8s集群时在想默认生成的证书有效期是多久，如下所示 /etc/kubernetes/pki/apiserver.crt …

Go语言 2023年5月25日
0062
微服务追踪SQL(支持Isto管控下的gorm查询追踪)

效果图 SQL的追踪正确插入到微服务的调用链之间详细记录了SQL的执行内容和消耗时间搜索SQL的类型多线程(goroutine)下的追踪效果在 Kubernetes 中部署…

Go语言 2023年5月25日
0075
Go 的 golang.org/x/ 系列包和标准库包有什么区别？

在开发过程中可能会遇到这样的情况，有一些包是引入自不同地方的，比如： golang.org/x/net/html 和 net/html， golang.org/x/crypto 和…

Go语言 2023年5月25日
0072
基于LSM的Key-Value数据库实现初篇

前篇文章对LSM的基本原理，算法流程做了简单的介绍，这篇文章将实现一个简单的基于LSM算法的迷你Key-Value数据库，结合上篇文章的理论与本篇文章的实践使之对LSM算法有更…

Go语言 2023年5月25日
0078
Golang项目的配置管理——Viper简易入门配置

Golang项目的配置管理——Viper简易入门配置 What is Viper？ From：https://github.com/spf13/viperViper is a co…

Go语言 2023年5月25日
0057
go-micro开发RPC服务的方法及其运行原理

go-micro是一个知名的golang微服务框架，最新版本是v4，这篇文章将介绍go-micro v4开发RPC服务的方法及其运作原理。基本概念 go-micro有几个重要的概…

Go语言 2023年5月25日
0062
开源项目｜Go 开发的一款分布式唯一 ID 生成系统

原文连接：开源项目｜Go 开发的一款分布式唯一 ID 生成系统今天跟大家介绍一个开源项目：id-maker，主要功能是用来在分布式环境下生成唯一 ID。上周停更了一周，也是用来…

Go语言 2023年5月25日
0076
Go内存逃逸分析

Go的内存逃逸及逃逸分析 Go的内存逃逸分析内存逃逸之前要搞清楚一件事我们编写的程序中的函数和局部变量默认是存放在栈上的(补充一点堆上存储的数据的指针是存放在栈上的因为…

Go语言 2023年5月25日
0052
基于Go语言的xmind读写库，我主要用来把有道云笔记思维导图转为xmind

xmind 基于go语言的xmind接口本库主要加载xmind文件为json结构,保存文件时也用的json结构而不是xml结构本库只做了最基本的主题添加功能,类似 &#…

Go语言 2023年5月25日
0076
websocket：二.Golang实现Websocket消息通知

我们在设计产品的时候通常都会遇到消息通知的时候，比如用户下单支付成功，比如用户有站内信来可以实时通知。而http是单向的，客户端请求，服务端返回，这次请求就已经结束。而websoc…

Go语言 2023年5月25日
0069
Maglev : A Fast and Reliable Software Network Load Balancer (using Consistent Hashing)

前言（为什么想读这一篇论文）这一篇论文吸引我注意的原因是，Consistent Hashing ;本来的特性就是作为分布式缓存之用。谷歌将他们的负载均衡器（代号：Maglev）发…

Go语言 2023年5月25日
0063
Go标准的目录结构（自总结）

微服务版 ├── LICENSE.md ├── …

Go语言 2023年5月25日
0055
基于知名微服务框架go-micro开发gRPC应用程序

go-micro是golang的一个微服务框架。 go-micro各个版本之间的兼容性问题一直被诟病，前几年go-micro更是分化出了两个分支：一个延续了go-micro，只不…

Go语言 2023年5月25日
00140
Go基础知识梳理（四）

Go基础知识梳理（四） GO的哲学是”不要通过共享内存来通信，而是通过通信来共享内存”，通道是GO通过通信来共享内存的载体。 rumtime包常用方法 ru…

Go语言 2023年5月25日
0049

2024 年 5 月
一	二	三	四	五	六	日
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30	31

Go 语言实现 gRPC 的发布订阅模式，REST 接口和超时控制

大家都在看