目录
前言
上篇文章 直接爆了，内容主要包括：简单的 grpc 服务，流处理模式，验证器，token 认证和证书认证。
在多个平台的阅读量都创了新高，在 oschina 更是获得了kb88凯时d88尊龙官网手机app官网登录首页推荐，阅读量到了 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 &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())
}
}()
<-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(), &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 上就有对应的输出了：
subtopic: golang: hello go
sub1: golang: hello go
sub1: docker: hello docker
也可以再多开几个订阅终端，那么每一个订阅终端上都会有相同的内容输出。
源码地址：
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 &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 都会设置一个超时时间，超过这个时间，如果还没有数据返回，服务端可能直接返回一个超时错误，或者客户端也可能结束这个连接。
如果没有这个超时时间，那是相当危险的。所有请求都阻塞在服务端，会消耗大量资源，比如内存。如果资源耗尽的话，甚至可能会导致整个服务崩溃。
那么，在 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, &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 := <-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() //超时后退出该go协程
case <-time.after(4 * time.second): // 模拟耗时操作
res := proto.helloreply{
message: "hello " req.name,
}
// //修改数据库前进行超时判断
// if ctx.err() == context.canceled{
// ...
// //如果已经超时，则退出
// }
data <- &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, &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
源码地址
总结
本文主要介绍了 grpc 的三部分实战内容，分别是：
- 发布订阅模式
- rest 接口
- 超时控制
个人感觉，超时控制还是最重要的，在平时的开发过程中需要多多注意。
结合上篇文章，grpc 的实战内容就写完了，代码全部可以执行，也都上传到了 github。
大家如果有任何疑问，欢迎给我留言，如果感觉不错的话，也欢迎关注和转发。
题图： 该图片由 reytschl 在 pixabay 上发布
源码地址：
推荐阅读
参考链接
更多关于grpc发布订阅rest接口的资料请关注其它相关文章！
