环境：
Golang： go1.18.2 windows/amd64
gRPC： v1.47.0
Protobuf： v1.28.0
完整代码：
https://github.com/WanshanTian/GolangLearning
cd GolangLearning/RPC/gRPC-Watch
1. 简介
客户端可以通过
Watch
机制来订阅服务器上某一节点的数据或状态，当其发生变化时可以收到相应的通知。
前阵子学习了
gRPC
的服务端流模式，今天我们就用这个流模式来具体实现Watch功能
2. 实践
现有下面一种场景：服务端保存着用户的年龄信息，客户端发送含用户姓名的message可以获取对应用户的年龄或者更新对应用户的年龄（年龄+1）；通过Watch功能可以实时监听用户年龄的状态，当有用户的年龄发生变化时，客户端收到通知
2.1 proto文件
2.1.1 新建gRPC-Watch文件夹，使用go mod init初始化，创建pb文件夹，新建query.proto文件
syntax = "proto3";
package pb;
option go_package= ".;pb";
import "google/protobuf/empty.proto";
// 定义查询服务包含的方法
service Query {
rpc GetAge (userInfo) returns (ageInfo) {};
rpc Update (userInfo) returns (google.protobuf.Empty) {};
rpc Watch (watchTime) returns (stream userInfo){}
}
// 请求用的结构体，包含一个name字段
message userInfo {
string name = 1;
}
// 响应用的结构体，包含一个age字段
message ageInfo {
int32 age = 1;
}
// watch的时间
message watchTime{
int32 time = 1;
}
-
GetAge
Update
-
Watch
- 当gRPC的方法不需要请求message或者不需要响应message时，可以先
import "google/protobuf/empty.proto"
google.protobuf.Empty
2.1.2 在.\gRPC-Watch\pb目录下使用protoc工具进行编译，在pb文件夹下直接生成.pb.go和_grpc.pb.go文件。关于protoc的详细使用可以查看【Golang | gRPC】使用protoc编译.proto文件
protoc --go_out=./ --go-grpc_out=./ .\query.proto
2.2 服务端
在gRPC-Watch目录下新建Server文件夹，新建main.go文件
2.2.1 下面我们通过Query这个结构体具体实现QueryServer接口
var userinfo = map[string]int32{
"foo": 18,
"bar": 20,
}
// Query 结构体，实现QueryServer接口
type Query struct {
mu sync.Mutex
ch chan string
pb.UnimplementedQueryServer // 涉及版本兼容
}
func (q *Query) GetAge(ctx context.Context, info *pb.UserInfo) (*pb.AgeInfo, error) {
age := userinfo[info.GetName()]
var res = new(pb.AgeInfo)
res.Age = age
return res, nil
}
//Update用于更新用户年龄，通过sync.Mutex加锁，如果年龄有更新，则向chan发送对应的用户名
func (q *Query) Update(ctx context.Context, info *pb.UserInfo) (*emptypb.Empty, error) {
q.mu.Lock()
defer q.mu.Unlock()
name := info.GetName()
userinfo[name] += 1
if q.ch != nil {
q.ch <- name
}
return &emptypb.Empty{}, nil
}
//Watch用于监听用户年龄状态的变化，先实例化一个chan，然后通过select方法监听chan内是否有数据，
//如果有则通过服务端流向客户端发送message，如果超过指定时间无更新，则退出
func (q *Query) Watch(timeSpecify *pb.WatchTime, serverStream pb.Query_WatchServer) error {
if q.ch != nil {
return errors.New("Watching is running, please stop first")
}
q.ch = make(chan string, 1)
for {
select {
case <-time.After(time.Second * time.Duration(timeSpecify.GetTime())):
close(q.ch)
q.ch = nil
return nil
case nameModify := <-q.ch:
log.Printf("The name of %s is updated\n", nameModify)
serverStream.Send(&pb.UserInfo{Name: nameModify})
}
}
}
-
Update
-
Watch
select
- 当Watch功能已经开启时，如果再次开启会返回报错
2.3 客户端
在gRPC-Watch目录下新建Client文件夹，新建main.go文件
func main() {
//建立无认证的连接
conn, err := grpc.Dial(":1234", grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
log.Panic(err)
}
defer conn.Close()
client := pb.NewQueryClient(conn)
//RPC方法调用
ctx := context.Background()
//先获取更新前的年龄
age, _ := client.GetAge(ctx, &pb.UserInfo{Name: "foo"})
log.Printf("Before updating, the age is %d\n", age.GetAge())
//更新年龄
log.Println("updating")
client.Update(ctx, &pb.UserInfo{Name: "foo"})
//再获取更新后的年龄
age, _ = client.GetAge(ctx, &pb.UserInfo{Name: "foo"})
log.Printf("After updating, the age is %d\n", age.GetAge())
}
2.4 Watch功能
在gRPC-Watch目录下新建Watch文件夹，新建main.go文件
func main() {
//建立无认证的连接
conn, err := grpc.Dial(":1234", grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
log.Panic(err)
}
defer conn.Close()
client := pb.NewQueryClient(conn)
//RPC方法调用
stream, _ := client.Watch(context.Background(), &pb.WatchTime{Time: 10})
for {
userInfoRecv, err := stream.Recv()
if err == io.EOF {
log.Println("end of watch")
break
} else if err != nil {
log.Println(err)
break
}
log.Printf("The name of %s is updated\n", userInfoRecv.GetName())
}
}
2.5 运行
首先开启服务端，然后开启Watch功能，在开启客户端，有如下输出结果（当指定时间内没有年龄更新，Watch自动退出）：
当Watch已经开启，并再次开启时，会返回如下自定义的报错，对应2.2代码里的
return errors.New("Watching is running, please stop first")
3. 总结
- Watch功能整体思路就是：服务端实例化一个chan，如果监听对象发生变化，向chan中发送值，客户端从chan中收到值，如果没有就一直阻塞