主要文件

/etcd/clientv3/watch.go

结构图

主要数据结构

//对外提供 Watch 接口，同时负责关闭 grpcStream 的流程。
type watcher struct {
//访问服务的RPC客户端remote   pb.WatchClient //etcdserver/etcdserverpb/rpc.pb.go...
// key=>grpcStream
streams map[string]*watchGrpcStreamlg      *zap.Logger
}

/*
1.实例化过程中，会创建 WatchClient, 这个 WatchClient 用于和服务端进行数据传输。同时会启动一个 goroutine 用于接收服务端的消息。
2.管理 watcherStream 关闭，创建等流程,
3.启动一个 goroutine 用于处理 reqc、respc 等管道的消息。
4.分发事件给所有监听者。
*/
type watchGrpcStream struct {owner    *watcherremote   pb.WatchClient
....//watchID=>watcherStream, 一个长连接substreams map[int64]*watcherStream//等待与服务器建立连接的 watcherStreamresuming []*watcherStream//客户端发起 watch 请求会被追加到管道reqc chan watchStreamRequest//服务端的响应追加到该管道respc chan *pb.WatchResponse....
resumec chan struct{}
}

/*
主要用于管理 watcheRequest 与各个管道之间的数据传输关系。
管道outc、recvc、buf 之间的关系是：(1)watchClient会有个goroutine 用于接收服务端的消息，并将这个消息追加到recvc 管道，(2)如果 recvc 中有数据，会读取出来然后追加到buf缓冲中。(3)如果 buf 不为空，会将消息发送给 outc, 客户端通过该管道，接收服务端的响应。
*/type watcherStream struct {//记录用户的请求实例initReq watchRequest//服务端响应会被追加到这个管道，outc chan WatchResponse//接收服务端的响应recvc chan *WatchResponse...//响应缓冲池buf []*WatchResponse
}

/*请求结构体
一个watcherRequest 代表一个监听请求。
*/
type watchRequest struct {ctx context.Context/*key 和 end 共同组成监控范围rev 是被监控的版本*/key stringend stringrev int64...//最终客户会通过该通道接收服务端的响应retc chan chan WatchResponse
}

//响应结构体
type WatchResponse struct {Header pb.ResponseHeader
Events []*EventCompactRevision int64
...
}

主要函数

监控key

/*
主要流程：
1.请求数据构建
2.生成 grpcStream 的在 streams 中的标识
3.如果标识不存在于 streams，创建一个 grpcStream
4.发起请求，等待请求响应
5.返回监听者接收消息的管道
*/
func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) WatchChan {var filters []pb.WatchCreateRequest_FilterType...//组装请求wr := &watchRequest{...}ok := false//获取streamKeyctxKey := streamKeyFromCtx(ctx)w.mu.Lock()if w.streams == nil { //streams 未初始化，有问题// closedw.mu.Unlock()ch := make(chan WatchResponse)close(ch)return ch}//如果已存在直接获取，不存则申请一个grpcStreamwgs := w.streams[ctxKey]if wgs == nil {wgs = w.newWatcherGrpcStream(ctx)w.streams[ctxKey] = wgs}...// couldn't create channel; return closed channelcloseCh := make(chan WatchResponse, 1)// 发送请求select {case reqc <- wr:ok = true....}//等待与服务器建立链接，同时获取接收消息的管道，并返回给监听者if ok {select {case ret := <-wr.retc:return ret...}...
}

创建GrpcStream 实例

func (w *watcher) newWatcherGrpcStream(inctx context.Context) *watchGrpcStream {ctx, cancel := context.WithCancel(&valCtx{inctx})wgs := &watchGrpcStream{...}go wgs.run()return wgs
}
/*
流程：
1.创建与服务交互的客户端实例
2.处理监听者请求
3.处理服务端消息
*/func (w *watchGrpcStream) run() {...defer func() {//关闭所有stream...}()//创建与服务端交互的客户端if wc, closeErr = w.newWatchClient(); closeErr != nil {return}...var cur *pb.WatchResponsefor {select {//监听请求case req := <-w.reqc:switch wreq := req.(type) {//监听请求case *watchRequest:outc := make(chan WatchResponse, 1)ws := &watcherStream{...}//启动一个 goroutine 用于处理 watcherStream 管道之间的数据传输go w.serveSubstream(ws, w.resumec)//等待与服务器建立连接的 ws//系统保证 watchRequest 会按照先进先出的规则与服务器建立连接w.resuming = append(w.resuming, ws)if len(w.resuming) == 1 { //如果只有一个，说明当前 ws 可以马上建立连接if err := wc.Send(ws.initReq.toPB()); err != nil {lg.Warningf("error when sending request: %v", err)}}//？？？？case *progressRequest:if err := wc.Send(wreq.toPB()); err != nil {lg.Warningf("error when sending request: %v", err)}}// 处理服务端的消息case pbresp := <-w.respc:...switch {case pbresp.Created:// response to head of queue creationif ws := w.resuming[0]; ws != nil {//注册 WathchId=>ws 的关系w.addSubstream(pbresp, ws)//通知所有的监听者w.dispatchEvent(pbresp)w.resuming[0] = nil}//处理下一个需要与服务器交互的请求if ws := w.nextResume(); ws != nil {if err := wc.Send(ws.initReq.toPB()); err != nil {...}}case pbresp.Canceled && pbresp.CompactRevision == 0:case cur.Fragment:continuedefault:}// watch client failed on Recv; spawn another if possiblecase err := <-w.errc:case <-w.ctx.Done():case ws := <-w.closingc:}}
}

watchStream 消息处理

/*
1.如果 buf 有消息，先发送到对应的管道
2.处理服务消息
(1)如果创建类型，需要看看是否给监听者发送过消息接收的管道，如果没有，就发送，同时设置已发送。
(2)更新下一次监听的版本号
(3)将需要发送给监听者的消息暂存到 buf 中
*/
func (w *watchGrpcStream) serveSubstream(ws *watcherStream, resumec chan struct{}) {//下一次监听的版本号nextRev := ws.initReq.rev...emptyWr := &WatchResponse{}for {curWr := emptyWroutc := ws.outcif len(ws.buf) > 0 { //如果缓冲中还有消息，先发送给监听者curWr = ws.buf[0]} else {outc = nil}select {case outc <- *curWr: //发送服务消息给监听者...case wr, ok := <-ws.recvc: //如果服务有消息发送过来if !ok {// shutdown from closeSubstreamreturn}//创建监听请求类型if wr.Created {if ws.initReq.retc != nil { //如果不为 nil,说明还没给监听者发送接收消息的 消息管道ws.initReq.retc <- ws.outc //给监听者发送接收消息管道ws.initReq.retc = nil //表示已经发送消息管道if ws.initReq.createdNotify { //发送消息该监听者ws.outc <- *wr}if ws.initReq.rev == 0 { //设置最新版本号nextRev = wr.Header.Revision}}} else { //更新下次接收变化的版本号nextRev = wr.Header.Revision}if wr.Created {continue}//将消息缓存到 buf 中ws.buf = append(ws.buf, wr)case <-w.ctx.Done():returncase <-ws.initReq.ctx.Done():returncase <-resumec:resuming = truereturn}}
}

消息分发

func (w *watchGrpcStream) dispatchEvent(pbresp *pb.WatchResponse) bool {events := make([]*Event, len(pbresp.Events))for i, ev := range pbresp.Events {events[i] = (*Event)(ev)}wr := &WatchResponse{...}if wr.IsProgressNotify() && pbresp.WatchId == -1 {return w.broadcastResponse(wr)}return w.unicastResponse(wr, pbresp.WatchId)}//将消息广播给所有监听者
func (w *watchGrpcStream) broadcastResponse(wr *WatchResponse) bool {for _, ws := range w.substreams {select {case ws.recvc <- wr:case <-ws.donec:}}return true
}
//将消息发送给特定监听者
func (w *watchGrpcStream) unicastResponse(wr *WatchResponse, watchId int64) bool {ws, ok := w.substreams[watchId]if !ok {return false}select {case ws.recvc <- wr:case <-ws.donec:return false}return true
}

客户端实例：
func (w *watchGrpcStream) newWatchClient() (pb.Watch_WatchClient, error) {...//打开 or 创建 客户端实例wc, err := w.openWatchClient() ...//goroutine 用于接收服务端消息go w.serveWatchClient(wc)return wc, nil
}func (w *watchGrpcStream) serveWatchClient(wc pb.Watch_WatchClient) {for {resp, err := wc.Recv()...select {case w.respc <- resp:...}
}

总结

1.监听通过 watcher、 watchGrpcStream 、watchStream、watcherRequest 和 WatchResponse 等数据结构实现，其中 wathcer 主要对外提供 Watch 接口和处理关闭流程，每一个 watcherRequest 会构建一个 watchGrpcStream，watchGrpcStream 用于管理 watchStream 和各种管道消息处理，watchStream 主要用于管理 watcheRequest 与各个管道之间的数据传输关系。

2.总共有3种后台 goroutine，watchGrpcStream.run 处理监听请求和服务端消息。 watchGrpcStream.serveSubstream(每个 watcherRequest 都会有)，作为服务端和监听者的中间层，将服务端的消息分发给监听者。watchGrpcStream.serveWatchClient 用于接收服务端的消息，并将消息发送给 w.respc, 等待 serveSubstream 处理。

3.Watch 主要实现是通过Stream 实现，只有其中的内部原理，有待研究。

ETCD 源码学习--Watch(client)相关推荐

ETCD 源码学习--lease(一)
在 ETCD 源码学习过程,不会讲解太多的源码知识,只讲解相关的实现机制,需要关注源码细节的朋友可以自行根据文章中的提示,找到相关源码进行学习. 租赁,在 etcd 中主要用于设置 key 的有效期, ...
ETCD 源码学习--WAL 实现
在 ETCD 源码学习过程,不会讲解太多的源码知识,只讲解相关的实现机制,需要关注源码细节的朋友可以自行根据文章中的提示,找到相关源码进行学习. 预写式日志(Write Ahead Log, WAL) ...
ETCD 源码学习--Raft 中 progress 的 inFlight 实现（九）
首先需要搞清什么是 inFlight,inFlight 在 Raft 中存储的是已发送给 Follower 的 MsgApp 消息,但没有收到 MsgAppResp 的消息 Index 值.简单的说 ...
菜鸟学源码之Nacos v1.1.3源码学习-Client模块(1):NacosNamingService初始化
摘要: 本文是Nacos源码学习的第一篇,基于Nacos v1.1.3版本对Nacos源码进行学习,本片主要从exmaple的App示例入手,切入Nacos客户端NacosNamingService的 ...
第十四课 k8s源码学习和二次开发原理篇-调度器原理
第十四课 k8s源码学习和二次开发原理篇-调度器原理 tags: k8s 源码学习 categories: 源码学习二次开发文章目录第十四课 k8s源码学习和二次开发原理篇-调度器原理第一节 ...
第八课 k8s源码学习和二次开发原理篇-KubeBuilder使用和Controller-runtime原理
第八课 k8s源码学习和二次开发原理篇-KubeBuilder使用和Controller-runtime原理 tags: k8s 源码学习 categories: 源码学习二次开发文章目录第八课 ...
Etcd源码分析-存储3
Etcd是分布式存储系统,当leader有数据变化,要及时更新到其他节点,这里就涉及到数据同步. 一.数据同步上一篇介绍,Etcd接收到客户端的请求,会把相关数据传递到Raft状态机中,那么进入状态 ...
client-go源码学习（二）：Reflector、DeltaFIFO
本文基于Kubernetes v1.22.4版本进行源码学习,对应的client-go版本为v0.22.4 3.Informer机制在Kubernetes系统中,组件之间通过HTTP协议进行通信,在 ...
第三课 k8s源码学习和二次开发-缓存机制Informers和Reflector组件学习
第三课 k8s源码学习和二次开发-缓存机制Informers和Reflector组件学习 tags: k8s 源码学习 categories: 源码学习二次开发文章目录第三课 k8s源码学习和二 ...

ETCD 源码学习--Watch(client)

主要文件

结构图

主要数据结构

主要函数

总结

ETCD 源码学习--Watch(client)相关推荐

最新文章

热门文章