1# dynamic/interface.go
 2
 3type ResourceInterface interface {
 4	Create(obj *unstructured.Unstructured, options metav1.CreateOptions, subresources ...string) (*unstructured.Unstructured, error)
 5	Update(obj *unstructured.Unstructured, options metav1.UpdateOptions, subresources ...string) (*unstructured.Unstructured, error)
 6	UpdateStatus(obj *unstructured.Unstructured, options metav1.UpdateOptions) (*unstructured.Unstructured, error)
 7	Delete(name string, options *metav1.DeleteOptions, subresources ...string) error
 8	DeleteCollection(options *metav1.DeleteOptions, listOptions metav1.ListOptions) error
 9	Get(name string, options metav1.GetOptions, subresources ...string) (*unstructured.Unstructured, error)
10	List(opts metav1.ListOptions) (*unstructured.UnstructuredList, error)
11	Watch(opts metav1.ListOptions) (watch.Interface, error)
12	Patch(name string, pt types.PatchType, data []byte, options metav1.PatchOptions, subresources ...string) (*unstructured.Unstructured, error)
13}

 1// DeploymentInterface has methods to work with Deployment resources.
 2type DeploymentInterface interface {
 3	Create(*v1.Deployment) (*v1.Deployment, error)
 4	Update(*v1.Deployment) (*v1.Deployment, error)
 5	UpdateStatus(*v1.Deployment) (*v1.Deployment, error)
 6	Delete(name string, options *metav1.DeleteOptions) error
 7	DeleteCollection(options *metav1.DeleteOptions, listOptions metav1.ListOptions) error
 8	Get(name string, options metav1.GetOptions) (*v1.Deployment, error)
 9	List(opts metav1.ListOptions) (*v1.DeploymentList, error)
10	Watch(opts metav1.ListOptions) (watch.Interface, error)
11	Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1.Deployment, err error)
12	GetScale(deploymentName string, options metav1.GetOptions) (*autoscalingv1.Scale, error)
13	UpdateScale(deploymentName string, scale *autoscalingv1.Scale) (*autoscalingv1.Scale, error)
14
15	DeploymentExpansion
16}

 1// client-go/tools/cache/reflector.go
 2type Reflector struct {
 3    // name identifies this reflector. By default it will be a file:line if possible.
 4    name string
 5    // metrics tracks basic metric information about the reflector
 6    metrics *reflectorMetrics// 但凡遇到metrics多半是用于做监控的，可以忽略
 7
 8    // The type of object we expect to place in the store.
 9    expectedType reflect.Type// 反射的类型，也就是要监控的对象类型，比如Pod
10    // The destination to sync up with the watch source
11    store Store// 存储，就是DeltaFIFO，为什么，后面会有代码证明
12    // listerWatcher is used to perform lists and watches.
13    listerWatcher ListerWatcher// 这个是用来从apiserver获取资源用的
14    // period controls timing between one watch ending and
15    // the beginning of the next one.
16    period       time.Duration// 反射器在List和Watch的时候理论上是死循环，只有出现错误才会退出
17    //period这个变量用在出错后多长时间再执行List和Watch，默认值是1秒钟
18    resyncPeriod time.Duration// 重新同步的周期，很多人肯定认为这个同步周期指的是从apiserver的同步周期
19    // resyncPeriod其实这里面同步指的是shared_informer使用者需要定期同步全量对象
20    ShouldResync func() bool// 如果需要同步，调用这个函数问一下，当然前提是该函数指针不为空
21    // clock allows tests to manipulate time
22    clock clock.Clock// 时钟
23    // lastSyncResourceVersion is the resource version token last
24    // observed when doing a sync with the underlying store
25    // it is thread safe, but not synchronized with the underlying store
26    lastSyncResourceVersion string// 最后一次同步的资源版本
27    // lastSyncResourceVersionMutex guards read/write access to lastSyncResourceVersion
28    lastSyncResourceVersionMutex sync.RWMutex// 还专门为最后一次同步的资源版本弄了个锁
29    // WatchListPageSize is the requested chunk size of initial and resync watch lists.
30    // Defaults to pager.PageSize.
31    WatchListPageSize int64
32}

  1// 代码源自client-go/tools/cache/reflector.go
  2func (r *Reflector) Run(stopCh <-chan struct{}) {
  3    // func Until(f func(), period time.Duration, stopCh <-chan struct{})是下面函数的声明
  4    // 这里面我们不用关心wait.Until是如何实现的，只要知道他调用函数f会被每period周期执行一次
  5    // 意思就是f()函数执行完毕再等period时间后在执行一次，也就是r.ListAndWatch()会被周期性的调用
  6    wait.Until(func() {
  7        if err := r.ListAndWatch(stopCh); err != nil {
  8            utilruntime.HandleError(err)
  9        }
 10    }, r.period, stopCh)
 11}
 12
 13// 代码源自client-go/tools/cache/reflector.go
 14func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
 15    var resourceVersion string
 16    // 很多存储类的系统都是这样设计的，数据采用版本的方式记录，数据每变化(添加、删除、更新)都会触发版本更新，
 17    // 这样的做法可以避免全量数据访问。以apiserver资源监控为例，只要监控比缓存中资源版本大的对象就可以了，
 18    // 把变化的部分更新到缓存中就可以达到与apiserver一致的效果，一般资源的初始版本为0，从0版本开始列举就是全量的对象了
 19    options := metav1.ListOptions{ResourceVersion: "0"}
 20    // 与监控相关的内容不多解释
 21    r.metrics.numberOfLists.Inc()
 22    start := r.clock.Now()
 23    // 列举资源，这部分是apimachery相关的内容，读者感兴趣可以自己了解
 24    list, err := r.listerWatcher.List(options)
 25    if err != nil {
 26        return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedType, err)
 27    }
 28    // 还是监控相关的
 29    r.metrics.listDuration.Observe(time.Since(start).Seconds())
 30    // 下面的代码主要是利用apimachinery相关的函数实现，就是把列举返回的结果转换为对象数组
 31    // 下面的代码大部分来自apimachinery，此处不做过多说明，读者只要知道实现什么功能就行了
 32    listMetaInterface, err := meta.ListAccessor(list)
 33    if err != nil {
 34        return fmt.Errorf("%s: Unable to understand list result %#v: %v", r.name, list, err)
 35    }
 36    resourceVersion = listMetaInterface.GetResourceVersion()
 37    
 38    items, err := meta.ExtractList(list)
 39    if err != nil {
 40        return fmt.Errorf("%s: Unable to understand list result %#v (%v)", r.name, list, err)
 41    }
 42    // 和监控相关的内容
 43    r.metrics.numberOfItemsInList.Observe(float64(len(items)))
 44    // 以上部分都是对象实例化的过程，可以称之为反射，也是Reflector这个名字的主要来源，本文不是讲解反射原理的，
 45    // 而是作为SharedInformer的前端，所以我们重点介绍的是对象在SharedInformer中流转过程，所以反射原理部分不做为重点讲解
 46    // 这可是真正从apiserver同步过来的全量对象，所以要同步到DeltaFIFO中
 47    if err := r.syncWith(items, resourceVersion); err != nil {
 48        return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
 49    }
 50    // 设置最新的同步的对象版本
 51    r.setLastSyncResourceVersion(resourceVersion)
 52    // 下面要启动一个后台协程实现定期的同步操作，这个同步就是将SharedInformer里面的对象全量以同步事件的方式通知使用者
 53    // 我们暂且称之为“后台同步协程”，Run()函数退出需要后台同步协程退出，所以下面的cancelCh就是干这个用的
 54    // 利用defer close(cancelCh)实现的，而resyncerrc是后台同步协程反向通知Run()函数的报错通道
 55    // 当后台同步协程出错，Run()函数接收到信号就可以退出了
 56    resyncerrc := make(chan error, 1)
 57    cancelCh := make(chan struct{})
 58    defer close(cancelCh)
 59    // 下面这个匿名函数就是后台同步协程的函数了
 60    go func() {
 61        // resyncCh返回的就是一个定时器，如果resyncPeriod这个为0那么就会返回一个永久定时器，cleanup函数是用来清理定时器的
 62        resyncCh, cleanup := r.resyncChan()
 63        defer func() {
 64            cleanup() 
 65        }()
 66        // 死循环等待各种信号
 67        for {
 68            // 只有定时器有信号才继续处理，其他的都会退出
 69            select {
 70            case <-resyncCh:
 71            case <-stopCh:
 72                return
 73            case <-cancelCh:
 74                return
 75            }
 76            // ShouldResync是个函数地址，创建反射器对象的时候传入，即便时间到了，也要通过函数问问是否需要同步
 77            if r.ShouldResync == nil || r.ShouldResync() {
 78                // 我们知道这个store是DeltaFIFO，DeltaFIFO.Resync()做了什么，读者自行温习相关的文章~
 79                // 就在这里实现了我们前面提到的同步，从这里看所谓的同步就是以全量对象同步事件的方式通知使用者
 80                if err := r.store.Resync(); err != nil {
 81                    resyncerrc <- err
 82                    return
 83                }
 84            }
 85            // 清理掉当前的计时器，获取下一个同步时间定时器
 86            cleanup()
 87            resyncCh, cleanup = r.resyncChan()
 88        }
 89    }()
 90 
 91    // 前面已经列举了全量对象，接下来就是watch的逻辑了
 92    for {
 93        // 如果有退出信号就立刻返回，否则就会往下走，因为有default.
 94        select {
 95        case <-stopCh:
 96            return nil
 97        default:
 98        }
 99 
100        // 计算watch的超时时间
101        timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
102        // 设置watch的选项，因为前期列举了全量对象，从这里只要监听最新版本以后的资源就可以了
103        // 如果没有资源变化总不能一直挂着吧？也不知道是卡死了还是怎么了，所以有一个超时会好一点
104        options = metav1.ListOptions{
105            ResourceVersion: resourceVersion,
106            TimeoutSeconds: &timeoutSeconds,
107        }
108        // 监控相关
109        r.metrics.numberOfWatches.Inc()
110        // 开始监控对象
111        w, err := r.listerWatcher.Watch(options)
112        // watch产生错误了，大部分错误就要退出函数然后再重新来一遍流程
113        if err != nil {
114            switch err {
115            case io.EOF:
116            case io.ErrUnexpectedEOF:
117            default:
118                utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedType, err))
119            }
120            // 类似于网络拒绝连接的错误要等一会儿再试，因为可能网络繁忙
121            if urlError, ok := err.(*url.Error); ok {
122                if opError, ok := urlError.Err.(*net.OpError); ok {
123                    if errno, ok := opError.Err.(syscall.Errno); ok && errno == syscall.ECONNREFUSED {
124                        time.Sleep(time.Second)
125                        continue
126                    }
127                }
128            }
129            return nil
130        }
131 
132        // watch返回是流，apiserver会将变化的资源通过这个流发送出来，client-go最终通过chan实现的
133        // 所以watchHandler()是一个需要持续从chan读取数据的流程，所以需要传入resyncerrc和stopCh
134        // 用于异步通知退出或者后台同步协程错误
135        if err := r.watchHandler(w, &resourceVersion, resyncerrc, stopCh); err != nil {
136            if err != errorStopRequested {
137                glog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
138            }
139            return nil
140        }
141    }
142}
143
144上面的函数中，调用了两个私有函数，分别为syncWith()和watchHandler()。syncWith()用于实现一次从apiserver全量对象的同步，这里的同步和我们上面提到的同步不是一回事，这里指的是从apiserver的同步。watchHandler是实现监控apiserver资源变化的处理过程，主要就是把apiserver的资源变化转换为DeltaFIFO调用。我们接下来就看这两个函数的具体实现
145
146接下来我们就要看看watchHandler做了什么？
147
148// 代码源自client-go/tools/cache/reflector.go
149// 实现apiserver全量对象的同步
150func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
151    // 做一次slice类型转换
152    found := make([]interface{}, 0, len(items))
153    for _, item := range items {
154        found = append(found, item)
155    }
156    // 直接调用了DeltaFIFO的Replace()接口，这个接口就是用于同步全量对象的
157    return r.store.Replace(found, resourceVersion)
158}
159// 实现从watch返回的chan中持续读取变化的资源，并转换为DeltaFIFO相应的调用
160func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
161    start := r.clock.Now()
162    eventCount := 0
163    // 监控相关
164    defer func() {
165        r.metrics.numberOfItemsInWatch.Observe(float64(eventCount))
166        r.metrics.watchDuration.Observe(time.Since(start).Seconds())
167    }()
168 
169    // 这里就开始无限循环的从chan中读取资源的变化，也可以理解为资源的增量变化，同时还要监控各种信号
170loop:
171    for {
172        select {
173        // 系统退出信号
174        case <-stopCh:
175            return errorStopRequested
176        // 后台同步协程出错信号
177        case err := <-errc:
178            return err
179        // watch函数返回的是一个chan，通过这个chan持续的读取对象
180        case event, ok := <-w.ResultChan():
181            // 如果不OK，说明chan关闭了，就要重新获取，这里面我们可以推测这个chan可能会运行过程中重新创建
182            // 否则就应该退出而不是继续循环
183            if !ok {
184                break loop
185            }
186            // 看来event可以作为错误的返回值，挺有意思，而不是通过关闭chan，这种方式可以传递错误信息，关闭chan做不到
187            if event.Type == watch.Error {
188                return apierrs.FromObject(event.Object)
189            }
190            // 这里面就是利用反射实例化对象了，而且判断了对象类型是我们设定的类型
191            if e, a := r.expectedType, reflect.TypeOf(event.Object); e != nil && e != a {
192                utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
193                continue
194            }
195            // 和list操作相似，也要获取对象的版本，要更新缓存中的版本，下次watch就可以忽略这些资源了
196            meta, err := meta.Accessor(event.Object)
197            if err != nil {
198                utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
199                continue
200            }
201            newResourceVersion := meta.GetResourceVersion()
202            // 根据事件的类型做不同的DeltaFIFO的操作
203            switch event.Type {
204            // 向DeltaFIFO添加一个添加的Delta
205            case watch.Added:
206                err := r.store.Add(event.Object)
207                if err != nil {
208                    utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err))
209                }
210            // 更新对象，向DeltaFIFO添加一个更新的Delta
211            case watch.Modified:
212                err := r.store.Update(event.Object)
213                if err != nil {
214                    utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", r.name, event.Object, err))
215            }
216            // 删除对象，向DeltaFIFO添加一个删除的Delta
217            case watch.Deleted:
218                err := r.store.Delete(event.Object)
219                if err != nil {
220                    utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err))
221            }
222            // 其他类型就不知道干什么了，只能报错
223            default:
224            utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
225            }
226            // 更新最新资源版本
227            *resourceVersion = newResourceVersion
228            r.setLastSyncResourceVersion(newResourceVersion)
229            eventCount++
230        }
231    }
232    // watch返回时间非常短而且没有任何事件要处理，这个属于异常现象，因为我们watch是设置了超时的
233    watchDuration := r.clock.Now().Sub(start)
234    if watchDuration < 1*time.Second && eventCount == 0 {
235        r.metrics.numberOfShortWatches.Inc()
236        return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
237    }
238 
239    return nil
240}
241
242// 代码源自client-go/tools/cache/reflector.go
243func (r *Reflector) setLastSyncResourceVersion(v string) {
244    // 设置已经获取到资源的最新版本
245    r.lastSyncResourceVersionMutex.Lock()
246    defer r.lastSyncResourceVersionMutex.Unlock()
247    r.lastSyncResourceVersion = v
248 
249    rv, err := strconv.Atoi(v)
250    if err == nil {
251        r.metrics.lastResourceVersion.Set(float64(rv))
252    }
253}
254 
255// 获取resync定时器，叫定时器比较好理解，叫chan很难和定时关联起来
256func (r *Reflector) resyncChan() (<-chan time.Time, func() bool) {
257    // 如果resyncPeriod说明就不用定时同步，返回的是永久超时的定时器
258    if r.resyncPeriod == 0 {
259        return neverExitWatch, func() bool { return false }
260    }
261    // 构建定时起
262    t := r.clock.NewTimer(r.resyncPeriod)
263    return t.C(), t.Stop
264}

1kubectl get pod podname -o yaml|grep resourceVersion

 1# tools/cache/delta_fifo.go
 2// DeltaType is the type of a change (addition, deletion, etc)
 3type DeltaType string // Delta的类型用字符串表达
 4
 5// Change type definition
 6const (
 7	Added   DeltaType = "Added"
 8	Updated DeltaType = "Updated"
 9	Deleted DeltaType = "Deleted"
10	// The other types are obvious. You'll get Sync deltas when:
11	//  * A watch expires/errors out and a new list/watch cycle is started.
12	//  * You've turned on periodic syncs.
13	// (Anything that trigger's DeltaFIFO's Replace() method.)
14	Sync DeltaType = "Sync" // 同步
15)
16
17// Delta is the type stored by a DeltaFIFO. It tells you what change
18// happened, and the object's state after* that change.
19//
20// [*] Unless the change is a deletion, and then you'll get the final
21//     state of the object before it was deleted.
22type Delta struct {
23	Type   DeltaType // Delta类型，比如增、减，后面有详细说明
24	Object interface{} // 对象，Delta的粒度是一个对象
25}
26
27// Deltas is a list of one or more 'Delta's to an individual object.
28// The oldest delta is at index 0, the newest delta is the last one.
29type Deltas []Delta // Delta数组

 1# tools/cache/delta_fifo.go
 2
 3// 这个接口类型就是下面面两个接口类型的组合了
 4// A KeyListerGetter is anything that knows how to list its keys and look up by key.
 5type KeyListerGetter interface {
 6	KeyLister
 7	KeyGetter
 8}
 9// 这是一个非常通用的接口类型，只定义了一个接口函数，就是返回所有的keys
10// A KeyLister is anything that knows how to list its keys.
11type KeyLister interface {
12	ListKeys() []string
13}
14// 这也是一个非常通用的接口类型，只定义了一个接口函数，就是通过key获取对象
15// A KeyGetter is anything that knows how to get the value stored under a given key.
16type KeyGetter interface {
17	GetByKey(key string) (interface{}, bool, error)
18}

 1# tools/cache/fifo.go
 2// 这个才是FIFO的抽象，DeltaFIFO只是FIFO的一种实现。
 3// Queue is exactly like a Store, but has a Pop() method too.
 4type Queue interface {
 5	Store // 实现了存储接口,这个很好理解，FIFO也是一种存储
 6    // 在存储的基础上增加了Pop接口，用于弹出对象
 7	// Pop blocks until it has something to process.
 8	// It returns the object that was process and the result of processing.
 9	// The PopProcessFunc may return an ErrRequeue{...} to indicate the item
10	// should be requeued before releasing the lock on the queue.
11	Pop(PopProcessFunc) (interface{}, error)
12    // 对象如果不在队列中就添加
13	// AddIfNotPresent adds a value previously
14	// returned by Pop back into the queue as long
15	// as nothing else (presumably more recent)
16	// has since been added.
17	AddIfNotPresent(interface{}) error
18    // 通过Replace()放入第一批对象到队列中并且已经被Pop()全部取走
19	// HasSynced returns true if the first batch of items has been popped
20	HasSynced() bool
21    // 关闭队列
22	// Close queue
23	Close()
24}
25

 1# tools/cache/delta_fifo.go
 2// DeltaFIFO is like FIFO, but allows you to process deletes.
 3//
 4// DeltaFIFO is a producer-consumer queue, where a Reflector is
 5// intended to be the producer, and the consumer is whatever calls
 6// the Pop() method.
 7//
 8// DeltaFIFO solves this use case:
 9//  * You want to process every object change (delta) at most once.
10//  * When you process an object, you want to see everything
11//    that's happened to it since you last processed it.
12//  * You want to process the deletion of objects.
13//  * You might want to periodically reprocess objects.
14//
15// DeltaFIFO's Pop(), Get(), and GetByKey() methods return
16// interface{} to satisfy the Store/Queue interfaces, but it
17// will always return an object of type Deltas.
18//
19// A note on threading: If you call Pop() in parallel from multiple
20// threads, you could end up with multiple threads processing slightly
21// different versions of the same object.
22//
23// A note on the KeyLister used by the DeltaFIFO: It's main purpose is
24// to list keys that are "known", for the purpose of figuring out which
25// items have been deleted when Replace() or Delete() are called. The deleted
26// object will be included in the DeleteFinalStateUnknown markers. These objects
27// could be stale.
28type DeltaFIFO struct {
29	// lock/cond protects access to 'items' and 'queue'.
30	lock sync.RWMutex// 读写锁，因为涉及到同时读写，读写锁性能要高
31	cond sync.Cond// 给Pop()接口使用，在没有对象的时候可以阻塞，内部锁复用读写锁
32    // 这个应该是Store的本质了，按照kv的方式存储对象，但是存储的是对象的Deltas数组
33	// We depend on the property that items in the set are in
34	// the queue and vice versa, and that all Deltas in this
35	// map have at least one Delta.
36	items map[string]Deltas
37	queue []string// 这个是为先入先出实现的，存储的就是对象的键
38    // 通过Replace()接口将第一批对象放入队列，或者第一次调用增、删、改接口时标记为true
39	// populated is true if the first batch of items inserted by Replace() has been populated
40	// or Delete/Add/Update was called first.
41	populated bool
42	// initialPopulationCount is the number of items inserted by the first call of Replace()
43	initialPopulationCount int// 通过Replace()接口将第一批对象放入队列的对象数量
44
45	// keyFunc is used to make the key used for queued item
46	// insertion and retrieval, and should be deterministic.
47	keyFunc KeyFunc// 对象键计算函数，在Indexer那篇文章介绍过
48    // 前面介绍就是为了这是用，该对象指向的就是Indexer，
49	// knownObjects list keys that are "known", for the
50	// purpose of figuring out which items have been deleted
51	// when Replace() or Delete() is called.
52	knownObjects KeyListerGetter
53
54	// Indication the queue is closed.
55	// Used to indicate a queue is closed so a control loop can exit when a queue is empty.
56	// Currently, not used to gate any of CRED operations.
57	closed     bool// 是否已经关闭的标记
58	closedLock sync.Mutex// 专为关闭设计的锁
59}
60
61var (
62	_ = Queue(&DeltaFIFO{}) // DeltaFIFO is a Queue
63)

 1// 代码源自client-go/tools/cache/delta_fifo.go
 2// 从函数名称来看把“动作”放入队列中，这个动作就是DeltaType，而且已经加锁了
 3func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
 4    // 前面提到的计算对象键的函数
 5    id, err := f.KeyOf(obj)
 6    if err != nil {
 7        return KeyError{obj, err}
 8    }
 9    // 如果是同步，并且对象未来会被删除，那么就直接返回，没必要记录这个动作了
10    // 肯定有人会问为什么Add/Delete/Update这些动作可以，因为同步对于已经删除的对象是没有意义的
11    // 已经删除的对象后续跟添加、更新有可能，因为同名的对象又被添加了，删除也是有可能
12    // 删除有些复杂，后面会有说明
13    if actionType == Sync && f.willObjectBeDeletedLocked(id) {
14        return nil
15    }
16    // 同一个对象的多次操作，所以要追加到Deltas数组中
17    newDeltas := append(f.items[id], Delta{actionType, obj})
18    // 合并操作，去掉冗余的delta
19    newDeltas = dedupDeltas(newDeltas)
20    // 判断对象是否已经存在
21    _, exists := f.items[id]
22    // 合并后操作有可能变成没有Delta么？后面的代码分析来看应该不会，所以暂时不知道这个判断目的
23    if len(newDeltas) > 0 {
24        // 如果对象没有存在过，那就放入队列中，如果存在说明已经在queue中了，也就没必要再添加了
25        if !exists {
26            f.queue = append(f.queue, id)
27        }
28        // 更新Deltas数组，通知所有调用Pop()的人
29        f.items[id] = newDeltas
30        f.cond.Broadcast()
31    } else if exists {
32        // 直接把对象删除，这段代码我不知道什么条件会进来，因为dedupDeltas()肯定有返回结果的
33        // 后面会有dedupDeltas()详细说明
34        delete(f.items, id)
35    }
36    return nil
37}

 1// 代码源自client-go/tools/cache/delta_fifo.go
 2func dedupDeltas(deltas Deltas) Deltas {
 3    // 小于2个delta，那就是1个呗，没啥好合并的
 4    n := len(deltas)
 5    if n < 2 {
 6        return deltas
 7    }
 8    // 取出最后两个
 9    a := &deltas[n-1]
10    b := &deltas[n-2]
11    // 判断如果是重复的，那就删除这两个delta把合并后的追加到Deltas数组尾部
12    if out := isDup(a, b); out != nil {
13        d := append(Deltas{}, deltas[:n-2]...)
14        return append(d, *out)
15    }
16    return deltas
17}
18// 判断两个Delta是否是重复的
19func isDup(a, b *Delta) *Delta {
20    // 只有一个判断，只能判断是否为删除类操作，和我们上面的判断相同
21    // 这个函数的本意应该还可以判断多种类型的重复，当前来看只能有删除这一种能够合并
22    if out := isDeletionDup(a, b); out != nil {
23        return out
24    }
25	
26    return nil
27}
28// 判断是否为删除类的重复
29func isDeletionDup(a, b *Delta) *Delta {
30    // 二者都是删除那肯定有一个是重复的
31    if b.Type != Deleted || a.Type != Deleted {
32        return nil
33    }
34    // 理论上返回最后一个比较好，但是对象已经不再系统监控范围，前一个删除状态是好的
35    if _, ok := b.Object.(DeletedFinalStateUnknown); ok {
36        return a
37    }
38    return b
39}

 1// 代码源自client-go/tools/cache/delta_fifo.go
 2func (f *DeltaFIFO) Replace(list []interface{}, resourceVersion string) error {
 3    f.lock.Lock()
 4    defer f.lock.Unlock()
 5    keys := make(sets.String, len(list))
 6    // 遍历所有的输入目标
 7    for _, item := range list {
 8        // 计算目标键
 9        key, err := f.KeyOf(item)
10        if err != nil {
11            return KeyError{item, err}
12        }
13        // 记录处理过的目标键，采用set存储，是为了后续快速查找
14        keys.Insert(key)
15        // 因为输入是目标全量，所以每个目标相当于重新同步了一次
16        if err := f.queueActionLocked(Sync, item); err != nil {
17            return fmt.Errorf("couldn't enqueue object: %v", err)
18        }
19    }
20    // 如果没有存储的话，自己存储的就是所有的老对象，目的要看看那些老对象不在全量集合中，那么就是删除的对象了
21    if f.knownObjects == nil {
22        // 遍历所有的元素
23        for k, oldItem := range f.items {
24            // 这个目标在输入的对象中存在就可以忽略
25            if keys.Has(k) {
26                continue
27            }
28            // 输入对象中没有，说明对象已经被删除了。
29            var deletedObj interface{}
30            if n := oldItem.Newest(); n != nil {
31                deletedObj = n.Object
32            }
33            // 终于看到哪里用到DeletedFinalStateUnknown了，队列中存储对象的Deltas数组中
34            // 可能已经存在Delete了，避免重复，采用DeletedFinalStateUnknown这种类型
35            if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
36                return err
37            }
38        }
39        
40        // 如果populated还没有设置，说明是第一次并且还没有任何修改操作执行过
41        if !f.populated {
42            f.populated = true
43            f.initialPopulationCount = len(list)  // 记录第一次通过来的对象数量
44        }
45 
46        return nil
47    }
48    // 下面处理的就是检测某些目标删除但是Delta没有在队列中
49    // 从存储中获取所有对象键
50    knownKeys := f.knownObjects.ListKeys()
51    queuedDeletions := 0
52    for _, k := range knownKeys {
53        // 对象还存在那就忽略
54        if keys.Has(k) {
55            continue
56        }
57        // 获取对象
58        deletedObj, exists, err := f.knownObjects.GetByKey(k)
59        if err != nil {
60            deletedObj = nil
61            glog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k)
62        } else if !exists {
63            deletedObj = nil
64            glog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k)
65        }
66        // 累积删除的对象数量
67        queuedDeletions++
68        // 把对象删除的Delta放入队列
69        if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil {
70            return err
71        }    
72    }
73    // 和上面的代码差不多，只是计算initialPopulationCount值的时候增加了删除对象的数量
74    if !f.populated {
75        f.populated = true
76        f.initialPopulationCount = len(list) + queuedDeletions
77    }
78 
79    return nil
80}

 1// 代码源自client-go/tools/cache/delta_fifo.go
 2func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
 3    f.lock.Lock()
 4    defer f.lock.Unlock()
 5    for {
 6        // 队列中有数据么？
 7        for len(f.queue) == 0 {
 8            // 看来是先判断的是否有数据，后判断是否关闭，这个和chan像
 9            if f.IsClosed() {
10                return nil, FIFOClosedError
11            }
12            // 没数据那就等待把
13            f.cond.Wait()
14        }
15        // 取出第一个对象
16        id := f.queue[0]
17        // 数组缩小，相当于把数组中的第一个元素弹出去了，这个不多解释哈
18        f.queue = f.queue[1:]
19        // 取出对象，因为queue中存的是对象键
20        item, ok := f.items[id]
21        // 同步对象计数减一，当减到0就说明外部已经全部同步完毕了
22        if f.initialPopulationCount > 0 {
23            f.initialPopulationCount--
24        }
25        // 对象不存在，这个是什么情况？貌似我们在合并对象的时候代码上有这个逻辑，估计永远不会执行
26        if !ok {
27            continue
28        }
29        // 把对象删除
30        delete(f.items, id)
31        // Pop()需要传入一个回调函数，用于处理对象
32        err := process(item)
33        // 如果需要重新入队列，那就重新入队列
34        if e, ok := err.(ErrRequeue); ok {
35            f.addIfNotPresent(id, item)
36            err = e.Err
37        }
38		
39        return item, err
40    }
41}

 1// 代码源自client-go/tools/cache/delta_fifo.go
 2func (f *DeltaFIFO) HasSynced() bool {
 3    f.lock.Lock()
 4    defer f.lock.Unlock()
 5    // 这里就比较明白了，一次同步全量对象后，并且全部Pop()出去才能算是同步完成
 6    // 其实这里所谓的同步就是全量内容已经进入Indexer，Indexer已经是系统中对象的全量快照了
 7    return f.populated && f.initialPopulationCount == 0
 8}
 9// 添加不存在的对象
10func (f *DeltaFIFO) AddIfNotPresent(obj interface{}) error {
11    // 这个要求放入的必须是Deltas数组，就是通过Pop（）弹出的对象
12    deltas, ok := obj.(Deltas)
13    if !ok {
14        return fmt.Errorf("object must be of type deltas, but got: %#v", obj)
15    }
16    // 多个Delta都是一个对象，所以用最新的就可以了
17    id, err := f.KeyOf(deltas.Newest().Object)
18    if err != nil {
19        return KeyError{obj, err}
20    }
21    // 后面有实现
22    f.lock.Lock()
23    defer f.lock.Unlock()
24    f.addIfNotPresent(id, deltas)
25    return nil
26}
27// 这个是添加不存在对象的实现
28func (f *DeltaFIFO) addIfNotPresent(id string, deltas Deltas) {
29    f.populated = true
30    // 这里判断的对象是否存在
31    if _, exists := f.items[id]; exists {
32        return
33    }
34    // 放入队列中
35    f.queue = append(f.queue, id)
36    f.items[id] = deltas
37    f.cond.Broadcast()
38}
39// 重新同步，这个在cache实现是空的，这里面有具体实现
40func (f *DeltaFIFO) Resync() error {
41    f.lock.Lock()
42    defer f.lock.Unlock()
43    // 如果没有Indexer那么重新同步是没有意义的，因为连同步了哪些对象都不知道
44    if f.knownObjects == nil {
45        return nil
46    }
47    // 列举Indexer里面所有的对象键
48    keys := f.knownObjects.ListKeys()
49    // 遍历对象键，为每个对象产生一个同步的Delta
50    for _, k := range keys {
51        // 具体实现后面有介绍
52        if err := f.syncKeyLocked(k); err != nil {
53            return err
54        }
55    }
56    return nil
57}
58// 具体对象同步实现接口
59func (f *DeltaFIFO) syncKeyLocked(key string) error {
60    // 获取对象
61    obj, exists, err := f.knownObjects.GetByKey(key)
62    if err != nil {
63        glog.Errorf("Unexpected error %v during lookup of key %v, unable to queue object for sync", err, key)
64        return nil
65    } else if !exists {
66        glog.Infof("Key %v does not exist in known objects store, unable to queue object for sync", key)
67        return nil
68    }
69    // 计算对象的键值，有人会问对象键不是已经传入了么？那个是存在Indexer里面的对象键，可能与这里的计算方式不同
70    id, err := f.KeyOf(obj)
71    if err != nil {
72        return KeyError{obj, err}
73    }
74    // 对象已经在存在，说明后续会通知对象的新变化，所以再加更新也没意义
75    if len(f.items[id]) > 0 {
76        return nil
77    }
78    // 添加对象同步的这个Delta
79    if err := f.queueActionLocked(Sync, obj); err != nil {
80        return fmt.Errorf("couldn't queue object: %v", err)
81    }
82    return nil
83}

1NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers) //NewDeltaFIFO
2---
3fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer) //sharedIndexInformer.Run

1// client-go/tools/cache/controller.go
2type ResourceEventHandler interface {
3    OnAdd(obj interface{})
4    OnUpdate(oldObj, newObj interface{})
5    OnDelete(obj interface{})
6}

 1// 代码源自client-go/tools/cache/shared_informer.go
 2type SharedInformer interface {
 3    // 添加资源事件处理器，关于ResourceEventHandler的定义在下面
 4    // 相当于注册回调函数，当有资源变化就会通过回调通知使用者，是不是能和上面介绍的Controller可以联系上了？
 5    // 为什么是Add不是Reg，说明可以支持多个handler
 6    AddEventHandler(handler ResourceEventHandler)
 7    // 上面添加的是不需要周期同步的处理器，下面的接口添加的是需要周期同步的处理器，周期同步上面提了好多遍了，不赘述
 8    AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration)
 9    // Store这个有专门的文章介绍，这个函数就是获取Store的接口,说明SharedInformer内有Store对象
10    GetStore() Store
11    // Controller在上面的章节介绍了，说明SharedInformer内有Controller对象
12    GetController() Controller
13    // 这个应该是SharedInformer的核心逻辑实现的地方
14    Run(stopCh <-chan struct{})
15    // 因为有Store，这个函数就是告知使用者Store里面是否已经同步了apiserver的资源，这个接口很有用
16    // 当创建完SharedInformer后，通过Reflector从apiserver同步全量对象，然后在通过DeltaFIFO一个一个的同志到cache
17    // 这个接口就是告知使用者，全量的对象是不是已经同步到了cache，这样就可以从cache列举或者查询了
18    HasSynced() bool
19    // 最新同步资源的版本，这个就不多说了，通过Controller(Controller通过Reflector)实现
20    LastSyncResourceVersion() string
21}
22// 扩展了SharedInformer类型，从类型名字上看共享的是Indexer，Indexer也是一种Store的实现
23type SharedIndexInformer interface {
24    // 继承了SharedInformer
25    SharedInformer
26    // 扩展了Indexer相关的接口
27    AddIndexers(indexers Indexers) error
28    GetIndexer() Indexer
29}
30// 代码源自client-go/tools/cache/controller.go，SharedInformer使用者如果需要处理资源的事件
31// 那么就要自己实现相应的回调函数
32type ResourceEventHandler interface {
33    // 添加对象回调函数
34    OnAdd(obj interface{})
35    // 更新对象回调函数
36    OnUpdate(oldObj, newObj interface{})
37    // 删除对象回调函数
38    OnDelete(obj interface{})
39}
40
41// 代码源自client-go/tools/cache/shared_informer.go
42type sharedIndexInformer struct {
43    // Indexer也是一种Store，这个我们知道的，Controller负责把Reflector和FIFO逻辑串联起来
44    // 所以这两个变量就涵盖了开篇那张图里面的Reflector、DeltaFIFO和LocalStore(cache)
45    indexer    Indexer
46    controller Controller
47    // sharedIndexInformer把上面提到的ResourceEventHandler进行了在层封装，并统一由sharedProcessor管理，后面章节专门介绍
48    processor             *sharedProcessor
49    // CacheMutationDetector其实没啥用，我理解是开发者自己实现的一个调试工具，用来发现对象突变的
50    // 实现方法也比较简单，DeltaFIFO弹出的对象在处理前先备份(深度拷贝)一份，然后定期比对两个对象是否相同
51    // 如果不同那就报警，说明处理过程中有人修改过对象，这个功能默认是关闭，所以我说没啥用
52    cacheMutationDetector CacheMutationDetector
53    // 这两个变量是给Reflector用的，我们知道Reflector是在Controller创建的
54    listerWatcher ListerWatcher
55    objectType    runtime.Object
56    // 定期同步的周期，因为可能存在多个ResourceEventHandler，就有可能存在多个同步周期，sharedIndexInformer采用最小的周期
57    // 这个周期值就存储在resyncCheckPeriod中，通过AddEventHandler()添加的处理器都采用defaultEventHandlerResyncPeriod
58    resyncCheckPeriod time.Duration
59    defaultEventHandlerResyncPeriod time.Duration
60    // 时钟
61    clock clock.Clock
62    // 启动、停止标记，肯定有人会问为啥用两个变量，一个变量不就可以实现启动和停止了么？
63    // 其实此处是三个状态，启动前，已启动和已停止，start表示了两个状态，而且为启动标记专门做了个锁
64    // 说明启动前和启动后有互斥的资源操作
65    started, stopped bool
66    startedLock      sync.Mutex
67 
68    // 这个名字起的也是够了，因为DeltaFIFO每次Pop()的时候需要传入一个函数用来处理Deltas
69    // 处理Deltas也就意味着要把消息通知给处理器，如果此时调用了AddEventHandler()
70    // 就会存在崩溃的问题，所以要有这个锁，阻塞Deltas....细想名字也没毛病~
71    blockDeltas sync.Mutex
72}

 1// 代码源自client-go/tools/cache/mutation_detector.go
 2// 默认关闭突变检测
 3var mutationDetectionEnabled = false
 4// 但是可以通过环境变量的KUBE_CACHE_MUTATION_DETECTOR开启
 5func init() {
 6    mutationDetectionEnabled, _ = strconv.ParseBool(os.Getenv("KUBE_CACHE_MUTATION_DETECTOR"))
 7}
 8 
 9// 这个是突变检测的类型抽象
10type CacheMutationDetector interface {
11    AddObject(obj interface{})  // 用于记录所有的对象
12    Run(stopCh <-chan struct{}) // 开启协程定期比对
13}
14// 创建CacheMutationDetector对象
15func NewCacheMutationDetector(name string) CacheMutationDetector {
16    // 如果没有开启选项就构造一个什么都不做的对象
17    if !mutationDetectionEnabled {
18        return dummyMutationDetector{}
19    }
20    // 如果开启了选项，那么就构造一个默认的突变检测器
21    glog.Warningln("Mutation detector is enabled, this will result in memory leakage.")
22    return &defaultCacheMutationDetector{name: name, period: 1 * time.Second}
23}
24// 这就是什么都不做的突变检测器
25type dummyMutationDetector struct{}
26func (dummyMutationDetector) Run(stopCh <-chan struct{}) {
27}
28func (dummyMutationDetector) AddObject(obj interface{}) {
29}

  1// 代码源自clien-go/tools/cache/shared_informer.go
  2type processorListener struct {
  3    // nextCh、addCh、handler、pendingNotifications的用法请参看我的《golang的chan有趣用法》里面有相关的例子
  4    // 总结这四个变量实现了事件的输入、缓冲、处理，事件就是apiserver资源的变化
  5    nextCh chan interface{}
  6    addCh  chan interface{}
  7    handler ResourceEventHandler
  8    pendingNotifications buffer.RingGrowing
  9    // 下面四个变量就是跟定时同步相关的了，requestedResyncPeriod是处理器设定的定时同步周期
 10    // resyncPeriod是跟sharedIndexInformer对齐的同步时间，因为sharedIndexInformer管理了多个处理器
 11    // 最终所有的处理器都会对齐到一个周期上，nextResync就是下一次同步的时间点
 12    requestedResyncPeriod time.Duration
 13    resyncPeriod time.Duration
 14    nextResync time.Time
 15    resyncLock sync.Mutex
 16}
 17
 18// 代码源自client-go/tools/cache/shared_informer.go
 19// 对，就这么简单，通过addCh传入，这里面的notification就是我们所谓的事件
 20func (p *processorListener) add(notification interface{}) {
 21    p.addCh <- notification
 22}
 23
 24// 代码源自client-go/tools/cache/shared_informer.go
 25// 这个函数是通过sharedProcessor利用wait.Group启动的，读者可以自行查看wait.Group
 26func (p *processorListener) pop() {
 27    defer utilruntime.HandleCrash()
 28    // nextCh是在这里，函数退出前析构的
 29    defer close(p.nextCh)
 30    // 临时变量，下面会用到
 31    var nextCh chan<- interface{}
 32    var notification interface{}
 33    // 进入死循环啦
 34    for {
 35        select {
 36        // 有两种情况，nextCh还没有初始化，这个语句就会被阻塞，这个我在《深入浅出golang之chan》说过
 37        // nextChan后面会赋值为p.nextCh，因为p.nextCh也是无缓冲的chan，数据不发送成功就阻塞                        
 38        case nextCh <- notification:
 39            // 如果发送成功了，那就从缓冲中再取一个事件出来
 40            var ok bool
 41            notification, ok = p.pendingNotifications.ReadOne()
 42            if !ok {
 43                // 如果没有事件，那就把nextCh再次设置为nil，接下来对于nextCh操作还会被阻塞
 44                nextCh = nil
 45            }
 46        // 从p.addCh读取一个事件出来，这回看到消费p.addCh的地方了
 47        case notificationToAdd, ok := <-p.addCh:
 48            // 说明p.addCh关闭了，只能退出
 49            if !ok {
 50                return
 51            }
 52            // notification为空说明当前还没发送任何事件给处理器
 53            if notification == nil {
 54                // 那就把刚刚获取的事件通过p.nextCh发送个处理器
 55                notification = notificationToAdd
 56                nextCh = p.nextCh
 57            } else {
 58                // 上一个事件还没有发送成功，那就先放到缓存中
 59                // pendingNotifications可以想象为一个slice，这样方便理解，是一个动态的缓存，
 60                p.pendingNotifications.WriteOne(notificationToAdd)
 61            }
 62        }
 63    }
 64}
 65pop()函数实现的非常巧妙，利用一个协程就把接收、缓冲、发送全部解决了。它充分的利用了golang的select可以同时操作多个chan的特性，同时从addChd读取数据从nextCh发送数据，这两个chan任何一个完成都可以激活协程。对于C/C++程序猿理解起来有点费劲，但这就是GO的魅力所在。接下来，我们看看从nextCh读取事件后是如何处理的：
 66
 67
 68// 代码源自client-go/tools/cache/shared_informer.go
 69// 这个也是sharedProcessor通过wait.Group启动的
 70func (p *processorListener) run() {
 71    // 因为wait.Until需要传入退出信号的chan
 72    stopCh := make(chan struct{})
 73    // wait.Until不多说了，我在前期不点的文章中说过了，只要没有收到退出信号就会周期的执行传入的函数
 74    wait.Until(func() {
 75        // wait.ExponentialBackoff()和wait.Until()类似，wait.Until()是无限循环
 76        // wait.ExponentialBackoff()是尝试几次，每次等待时间会以指数上涨
 77        err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
 78            // 这也是chan的range用法，可以参看我的《深入浅出golang的chan》了解细节
 79            for next := range p.nextCh {
 80                // 判断事件类型，这里面的handler就是调用SharedInfomer.AddEventHandler()传入的
 81                // 理论上处理的不是Deltas么？怎么变成了其他类型，这是SharedInformer做的二次封装，后面会看到
 82                switch notification := next.(type) {
 83                case updateNotification:
 84                    p.handler.OnUpdate(notification.oldObj, notification.newObj)
 85                case addNotification:
 86                    p.handler.OnAdd(notification.newObj)
 87                case deleteNotification:
 88                    p.handler.OnDelete(notification.oldObj)
 89                default:
 90                    utilruntime.HandleError(fmt.Errorf("unrecognized notification: %#v", next))
 91                }
 92            }
 93 
 94            return true, nil
 95        })
 96 
 97        // 执行到这里只能是nextCh已经被关闭了，所以关闭stopCh,通知wait.Until()退出
 98        if err == nil {
 99            close(stopCh)
100        }
101    }, 1*time.Minute, stopCh)
102}

 1// client-go/tools/cache/shared_informer.go
 2// sharedProcessor是通过数组组织处理器的，只是分了需要定时同步和不需要要同步两类
 3type sharedProcessor struct {
 4    listenersStarted bool                 // 所有处理器是否已经启动的标识
 5    listenersLock    sync.RWMutex         // 读写锁
 6    listeners        []*processorListener // 通用的处理器
 7    syncingListeners []*processorListener // 需要定时同步的处理器
 8    clock            clock.Clock          // 时钟
 9    wg               wait.Group           // 前面讲过了processorListener每个需要两个协程，
10                                          // 用wait.Group来管理所有处理器的携程，保证他们都能退出
11}
12// 代码源自client-go/tools/cache/shared_informer.go
13// 添加处理器，sharedIndexInformer.AddEventHandler()就会调用这个函数实现处理器的添加
14func (p *sharedProcessor) addListener(listener *processorListener) {
15    // 加锁，这个很好理解
16    p.listenersLock.Lock()
17    defer p.listenersLock.Unlock()
18    // 把处理器添加到数组中
19    p.addListenerLocked(listener)
20    // 通过wait.Group启动两个协程，做的事情我们在processorListener说过了，这里就是我们上面提到的启动两个协程的地方
21    // 这个地方判断了listenersStarted，这说明sharedProcessor在启动前、后都可以添加处理器
22    if p.listenersStarted {
23        p.wg.Start(listener.run)
24        p.wg.Start(listener.pop)
25    }
26}
27// 把处理器添加到数组中
28func (p *sharedProcessor) addListenerLocked(listener *processorListener) {
29    // 两类(定时同步和不同步)的处理器数组都添加了，这是因为没有定时同步的也会用默认的时间，后面我们会看到
30    // 那么问题来了，那还用两个数组干什么呢？
31    p.listeners = append(p.listeners, listener)
32    p.syncingListeners = append(p.syncingListeners, listener)
33}
34
35// 代码源自client-go/tools/cache/shared_informer.go
36// 通过函数名称也能感觉到分发的感觉~sync表示obj对象是否为同步事件对象
37func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
38    // 加锁没毛病
39    p.listenersLock.RLock()
40    defer p.listenersLock.RUnlock()
41 
42    // 无论是否为sync，添加处理器的代码中我们知道两个数组都会被添加，所以判断不判断没啥区别~
43    // 所以我的猜测是代码以前实现的是明显区分两类的，但随着代码的更新二者的界限已经没那么明显了
44    if sync {
45        for _, listener := range p.syncingListeners {
46            listener.add(obj)
47        }
48    } else {
49        for _, listener := range p.listeners {
50            listener.add(obj)
51        }
52    }
53}
54// 代码源自client-go/tools/cache/shared_informer.go
55func (p *sharedProcessor) run(stopCh <-chan struct{}) {
56    // 启动前、后对于添加处理器的逻辑是不同，启动前的处理器是不会立刻启动连个协程执行处理器的pop()和run()函数的
57    // 而是在这里统一的启动
58    func() {
59        p.listenersLock.RLock()
60        defer p.listenersLock.RUnlock()
61        // 遍历所有的处理器，然后为处理器启动两个后台协程
62        for _, listener := range p.listeners {
63            p.wg.Start(listener.run)
64            p.wg.Start(listener.pop)
65        }
66        p.listenersStarted = true
67	}()
68    // 等待退出信号
69    <-stopCh
70    p.listenersLock.RLock()
71    defer p.listenersLock.RUnlock()
72    // 关闭addCh，processorListener.pop()这个协程就会退出，不明白的可以再次回顾代码
73    // 因为processorListener.pop()会关闭processorListener.nextCh，processorListener.run()就会退出
74    // 所以这里只要关闭processorListener.addCh就可以自动实现两个协程的退出，不得不说设计的还是挺巧妙的
75    for _, listener := range p.listeners {
76        close(listener.addCh) 
77    }
78    // 等待所有的协程退出，这里指的所有协程就是所有处理器的那两个协程
79    p.wg.Wait()
80}

  1// 代码源自client-go/tools/cache/shared_informer.go
  2// lw:这个是apiserver客户端相关的，用于Reflector从apiserver获取资源，所以需要外部提供
  3// objType:这个SharedInformer监控的对象类型
  4// resyncPeriod:同步周期，SharedInformer需要多长时间给使用者发送一次全量对象的同步时间
  5func NewSharedInformer(lw ListerWatcher, objType runtime.Object, resyncPeriod time.Duration) SharedInformer {
  6    // 还是用SharedIndexInformer实现的
  7    return NewSharedIndexInformer(lw, objType, resyncPeriod, Indexers{})
  8}
  9// 创建SharedIndexInformer对象，其中大部分参数再上面的函数已经介绍了
 10// indexers:需要外部提供计算对象索引键的函数，也就是这里面的对象需要通过什么方式创建索引
 11func NewSharedIndexInformer(lw ListerWatcher, objType runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {
 12    realClock := &clock.RealClock{}
 13    sharedIndexInformer := &sharedIndexInformer{
 14        // 管理所有处理器用的，这个上面的章节解释了
 15        processor:                       &sharedProcessor{clock: realClock},
 16        // 其实就是在构造cache，读者可以自行查看NewIndexer()的实现，
 17        // 在cache中的对象用DeletionHandlingMetaNamespaceKeyFunc计算对象键，用indexers计算索引键
 18        // 可以想象成每个对象键是Namespace/Name，每个索引键是Namespace，即按照Namesapce分类
 19        // 因为objType决定了只有一种类型对象，所以Namesapce是最大的分类
 20        indexer:                         NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),
 21        // 下面这两主要就是给Controller用，确切的说是给Reflector用的
 22        listerWatcher:                   lw,
 23        objectType:                      objType,
 24        // 无论是否需要定时同步，SharedInformer都提供了一个默认的同步时间，当然这个是外部设置的
 25        resyncCheckPeriod:               defaultEventHandlerResyncPeriod,
 26        defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
 27        // 默认没有开启的对象突变检测器，没啥用，也不多介绍
 28        cacheMutationDetector:           NewCacheMutationDetector(fmt.Sprintf("%T", objType)),
 29        clock: realClock,
 30    }
 31    return sharedIndexInformer
 32}
 33// 代码源自client-go/tools/cache/shared_informer.go
 34// 添加没有指定同步周期的事件处理器
 35func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) {
 36    // defaultEventHandlerResyncPeriod是默认的同步周期，在创建SharedInformer的时候设置的
 37    s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
 38}
 39// 添加需要定期同步的事件处理器
 40func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) {
 41    // 因为是否已经开始对于添加事件处理器的方式不同，后面会有介绍，所以此处加了锁
 42    s.startedLock.Lock()
 43    defer s.startedLock.Unlock()
 44 
 45    // 如果已经结束了，那就可以直接返回了
 46    if s.stopped {
 47        return
 48    }
 49    // 如果有同步周期，==0就是永远不用同步
 50    if resyncPeriod > 0 {
 51        // 同步周期不能太短，太短对于系统来说反而是个负担，大量的无效计算浪费在这上面
 52        if resyncPeriod < minimumResyncPeriod {
 53            resyncPeriod = minimumResyncPeriod
 54        }
 55        // SharedInformer管理了很多处理器，每个处理器都有自己的同步周期，所以此处要统一成一个，称之为对齐
 56        // SharedInformer会选择所有处理器中最小的那个作为所有处理器的同步周期，称为对齐后的同步周期
 57        // 此处就要判断是不是比当前对齐后的同步周期还要小
 58        if resyncPeriod < s.resyncCheckPeriod {
 59            // 如果已经启动了，那么只能用和大家一样的周期
 60            if s.started {
 61                resyncPeriod = s.resyncCheckPeriod
 62            // 如果没启动，那就让大家都用最新的对齐同步周期
 63            } else {
 64                s.resyncCheckPeriod = resyncPeriod
 65                s.processor.resyncCheckPeriodChanged(resyncPeriod)
 66            }
 67        }
 68    }
 69    // 创建处理器，代码一直用listener,可能想强调没事件就挂起把，我反而想用处理器这个名词
 70    // determineResyncPeriod()这个函数读者自己分析把，非常简单，这里面只要知道创建了处理器就行了
 71    listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize)
 72    // 如果没有启动，那么直接添加处理器就可以了
 73    if !s.started {
 74        s.processor.addListener(listener)
 75        return
 76    }
 77 
 78    // 这个锁就是暂停再想所有的处理器分发事件用的，因为这样会遍历所有的处理器，此时添加会有风险
 79    s.blockDeltas.Lock()
 80    defer s.blockDeltas.Unlock()
 81    // 添加处理器
 82    s.processor.addListener(listener)
 83    // 这里有意思啦，遍历缓冲中的所有对象，通知处理器，因为SharedInformer已经启动了，可能很多对象已经让其他的处理器处理过了，
 84    // 所以这些对象就不会再通知新添加的处理器，此处就是解决这个问题的
 85    for _, item := range s.indexer.List() {
 86        listener.add(addNotification{newObj: item})
 87    }
 88}
 89// 代码源自client-go/tools/cache/shared_informer.go
 90// sharedIndexInformer的核心逻辑函数
 91func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
 92    defer utilruntime.HandleCrash()
 93    // 在此处构造的DeltaFIFO
 94    fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)
 95    // 这里的Config是我们介绍Reflector时介绍的那个Config
 96    cfg := &Config{
 97        // 我前面一直在说Reflector输入到DeltaFIFO,这里算是直接证明了
 98        Queue:            fifo,            
 99        // 下面这些变量我们在Reflector都说了，这里赘述
100        ListerWatcher:    s.listerWatcher, 
101        ObjectType:       s.objectType,
102        FullResyncPeriod: s.resyncCheckPeriod,
103        RetryOnError:     false,
104        ShouldResync:     s.processor.shouldResync,
105        // 这个才是重点，Controller调用DeltaFIFO.Pop()接口传入的就是这个回调函数，也是我们后续重点介绍的
106        Process: s.HandleDeltas,
107    }
108    // 创建Controller，这个不用多说了
109    func() {
110        s.startedLock.Lock()
111        defer s.startedLock.Unlock()
112 
113        s.controller = New(cfg)
114        s.controller.(*controller).clock = s.clock
115        s.started = true
116    }()
117    // 这个processorStopCh 是给sharedProcessor和cacheMutationDetector传递退出信号的
118    // 因为这里要创建两个协程运行sharedProcessor和cacheMutationDetector的核心函数
119    processorStopCh := make(chan struct{})
120    var wg wait.Group
121    defer wg.Wait()              // Wait for Processor to stop
122    defer close(processorStopCh) // Tell Processor to stop
123    wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
124    wg.StartWithChannel(processorStopCh, s.processor.run)
125 
126    // Run()函数都退出了，也就应该设置结束的标识了
127    defer func() {
128        s.startedLock.Lock()
129        defer s.startedLock.Unlock()
130        s.stopped = true 
131    }()
132    // 启动Controller，Controller一旦运行，整个流程就开始启动了，所以叫Controller也不为过
133    // 毕竟Controller是SharedInformer的发动机嘛
134    s.controller.Run(stopCh)
135}
136
137sharedIndexInformer通过Run()函数启动了Controller和sharedProcess()，Controller通过DeltaFIFO.Pop()函数弹出Deltas，并调用函数处理，这个处理函数就是sharedIndexInformer.HandleDeltas()，这个函数是衔接Controller和sharedProcess的关键点，他把Deltas转换为sharedProcess需要的各种Notification类型。下面我们就对这个函数进行代码分析：
138
139// 代码源自client-go/tools/cache/shared_informer.go
140func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
141    // 看到这里就知道为啥起名为blockDeltas了，这是阻塞处理器Deltas啊~因为分发事件到处理器，所以要加锁
142    s.blockDeltas.Lock()
143    defer s.blockDeltas.Unlock()
144 
145    // Deltas里面包含了一个对象的多个增量操作，所以要从最老的Delta到最先的Delta遍历处理
146    for _, d := range obj.(Deltas) {
147        // 根据不同的Delta做不同的操作，但是大致分为对象添加、删除两大类操作
148        // 所有的操作都要先同步到cache在通知处理器，这样保持处理器和cache的状态是一致的
149        switch d.Type {
150        // 同步、添加、更新都是对象添加类的造作，至于是否是更新还要看cache是否有这个对象
151        case Sync, Added, Updated:
152            // 看看对象是不是有定时同步产生的事件
153            isSync := d.Type == Sync
154            // 检测突变，没啥用
155            s.cacheMutationDetector.AddObject(d.Object)
156            // 如果cache中有的对象，一律看做是更新事件
157            if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
158                // 把对象更新到cache中
159                if err := s.indexer.Update(d.Object); err != nil {
160                    return err
161                }
162                // 通知处理器处理事件
163                s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
164            // cache中没有的对象，一律看做是新增事件
165            } else {
166                // 把对象添加到cache中
167                if err := s.indexer.Add(d.Object); err != nil {
168                    return err
169                }
170                // 通知处理器处理器事件
171                s.processor.distribute(addNotification{newObj: d.Object}, isSync)
172            }
173        // 对象被删除
174        case Deleted:
175            // 从cache中删除对象
176            if err := s.indexer.Delete(d.Object); err != nil {
177                return err
178            }
179            // 通知所有的处理器对象被删除了
180            s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
181        }
182    }
183    return nil
184}

 1// client-go/tools/cache/listwatch.go
 2type Lister interface {
 3    // List should return a list type object; the Items field will be extracted, and the
 4    // ResourceVersion field will be used to start the watch in the right place.
 5    List(options metav1.ListOptions) (runtime.Object, error)
 6}
 7
 8// Watcher is any object that knows how to start a watch on a resource.
 9type Watcher interface {
10    // Watch should begin a watch at the specified version.
11    Watch(options metav1.ListOptions) (watch.Interface, error)
12}
13
14// ListerWatcher is any object that knows how to perform an initial list and start a watch on a resource.
15type ListerWatcher interface {
16    Lister
17    Watcher
18}

1// client-go/examples/workqueue/main.go
2// create the pod watcher
3podListWatcher := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), "pods", v1.NamespaceDefault, fields.Everything())

 1/ k8s.io/api/core/v1/types.go
 2const (
 3    // Pods, number
 4    ResourcePods ResourceName = "pods"
 5    // Services, number
 6    ResourceServices ResourceName = "services"
 7    // ReplicationControllers, number
 8    ResourceReplicationControllers ResourceName = "replicationcontrollers"
 9    // ResourceQuotas, number
10    ResourceQuotas ResourceName = "resourcequotas"
11    // ResourceSecrets, number
12    ResourceSecrets ResourceName = "secrets"
13    // ResourceConfigMaps, number
14    ResourceConfigMaps ResourceName = "configmaps"
15    // ResourcePersistentVolumeClaims, number
16    ResourcePersistentVolumeClaims ResourceName = "persistentvolumeclaims"
17    // ResourceServicesNodePorts, number
18    ResourceServicesNodePorts ResourceName = "services.nodeports"
19    // ResourceServicesLoadBalancers, number
20    ResourceServicesLoadBalancers ResourceName = "services.loadbalancers"
21    // CPU request, in cores. (500m = .5 cores)
22    ResourceRequestsCPU ResourceName = "requests.cpu"
23    // Memory request, in bytes. (500Gi = 500GiB = 500 * 1024 * 1024 * 1024)
24    ResourceRequestsMemory ResourceName = "requests.memory"
25    // Storage request, in bytes
26    ResourceRequestsStorage ResourceName = "requests.storage"
27    // Local ephemeral storage request, in bytes. (500Gi = 500GiB = 500 * 1024 * 1024 * 1024)
28    ResourceRequestsEphemeralStorage ResourceName = "requests.ephemeral-storage"
29    // CPU limit, in cores. (500m = .5 cores)
30    ResourceLimitsCPU ResourceName = "limits.cpu"
31    // Memory limit, in bytes. (500Gi = 500GiB = 500 * 1024 * 1024 * 1024)
32    ResourceLimitsMemory ResourceName = "limits.memory"
33    // Local ephemeral storage limit, in bytes. (500Gi = 500GiB = 500 * 1024 * 1024 * 1024)
34    ResourceLimitsEphemeralStorage ResourceName = "limits.ephemeral-storage"
35)

 1// client-go/kubernetes/clientset.go
 2type Clientset struct {
 3    *discovery.DiscoveryClient
 4    admissionregistrationV1beta1 *admissionregistrationv1beta1.AdmissionregistrationV1beta1Client
 5    appsV1                       *appsv1.AppsV1Client
 6    appsV1beta1                  *appsv1beta1.AppsV1beta1Client
 7    appsV1beta2                  *appsv1beta2.AppsV1beta2Client
 8    auditregistrationV1alpha1    *auditregistrationv1alpha1.AuditregistrationV1alpha1Client
 9    authenticationV1             *authenticationv1.AuthenticationV1Client
10    authenticationV1beta1        *authenticationv1beta1.AuthenticationV1beta1Client
11    authorizationV1              *authorizationv1.AuthorizationV1Client
12    authorizationV1beta1         *authorizationv1beta1.AuthorizationV1beta1Client
13    autoscalingV1                *autoscalingv1.AutoscalingV1Client
14    autoscalingV2beta1           *autoscalingv2beta1.AutoscalingV2beta1Client
15    autoscalingV2beta2           *autoscalingv2beta2.AutoscalingV2beta2Client
16    batchV1                      *batchv1.BatchV1Client
17    batchV1beta1                 *batchv1beta1.BatchV1beta1Client
18    batchV2alpha1                *batchv2alpha1.BatchV2alpha1Client
19    certificatesV1beta1          *certificatesv1beta1.CertificatesV1beta1Client
20    coordinationV1beta1          *coordinationv1beta1.CoordinationV1beta1Client
21    coordinationV1               *coordinationv1.CoordinationV1Client
22    coreV1                       *corev1.CoreV1Client
23    eventsV1beta1                *eventsv1beta1.EventsV1beta1Client
24    extensionsV1beta1            *extensionsv1beta1.ExtensionsV1beta1Client
25    networkingV1                 *networkingv1.NetworkingV1Client
26    networkingV1beta1            *networkingv1beta1.NetworkingV1beta1Client
27    nodeV1alpha1                 *nodev1alpha1.NodeV1alpha1Client
28    nodeV1beta1                  *nodev1beta1.NodeV1beta1Client
29    policyV1beta1                *policyv1beta1.PolicyV1beta1Client
30    rbacV1                       *rbacv1.RbacV1Client
31    rbacV1beta1                  *rbacv1beta1.RbacV1beta1Client
32    rbacV1alpha1                 *rbacv1alpha1.RbacV1alpha1Client
33    schedulingV1alpha1           *schedulingv1alpha1.SchedulingV1alpha1Client
34    schedulingV1beta1            *schedulingv1beta1.SchedulingV1beta1Client
35    schedulingV1                 *schedulingv1.SchedulingV1Client
36    settingsV1alpha1             *settingsv1alpha1.SettingsV1alpha1Client
37    storageV1beta1               *storagev1beta1.StorageV1beta1Client
38    storageV1                    *storagev1.StorageV1Client
39    storageV1alpha1              *storagev1alpha1.StorageV1alpha1Client
40}

 1// client-go/rest/client.go
 2type Interface interface {
 3    GetRateLimiter() flowcontrol.RateLimiter
 4    Verb(verb string) *Request
 5    Post() *Request
 6    Put() *Request
 7    Patch(pt types.PatchType) *Request
 8    Get() *Request
 9    Delete() *Request
10    APIVersion() schema.GroupVersion
11}

 1// 代码源自client-go/tools/cache/controller.go
 2type Config struct {
 3    Queue                          // SharedInformer使用DeltaFIFO
 4    ListerWatcher                  // 这个用来构造Reflector
 5    Process ProcessFunc            // 这个在调用DeltaFIFO.Pop()使用，弹出对象要如何处理
 6    ObjectType runtime.Object      // 对象类型，这个肯定是Reflector使用
 7    FullResyncPeriod time.Duration // 全量同步周期，这个在Reflector使用
 8    ShouldResync ShouldResyncFunc  // Reflector在全量更新的时候会调用该函数询问
 9    RetryOnError bool              // 错误是否需要尝试
10}
11
12
13
14// 这是一个Controller的抽象
15type Controller interface {
16    Run(stopCh <-chan struct{})      // 核心流程函数
17    HasSynced() bool                 // apiserver中的对象是否已经同步到了Store中
18    LastSyncResourceVersion() string // 最新的资源版本号
19}
20
21// 代码源自client-go/tools/cache/controller.go
22// controller是Controller的实现类型
23type controller struct {
24    config         Config       // 配置，上面有讲解
25    reflector      *Reflector   // 反射器
26    reflectorMutex sync.RWMutex // 反射器的锁
27    clock          clock.Clock  // 时钟
28}
29// 核心业务逻辑实现
30func (c *controller) Run(stopCh <-chan struct{}) {
31    defer utilruntime.HandleCrash()
32    // 创建一个协程，如果收到系统退出的信号就关闭队列，相当于在这里析构的队列
33    go func() {
34        <-stopCh
35        c.config.Queue.Close()
36    }()
37    // 创建Reflector，传入的参数都是我们上一个章节解释过的，这里不赘述
38    r := NewReflector(
39        c.config.ListerWatcher,
40        c.config.ObjectType,
41        c.config.Queue,
42        c.config.FullResyncPeriod,
43    )
44    // r.ShouldResync的存在就是为了以后使用少些一点代码？否则直接使用c.config.ShouldResync不就完了么？不明白用意
45    r.ShouldResync = c.config.ShouldResync
46    r.clock = c.clock
47    // 记录反射器
48    c.reflectorMutex.Lock()
49    c.reflector = r
50    c.reflectorMutex.Unlock()
51    // wait.Group不是本章的讲解内容，只要把它理解为类似barrier就行了
52    // 被他管理的所有的协程都退出后调用Wait()才会退出，否则就会被阻塞
53    var wg wait.Group
54    defer wg.Wait()
55    // StartWithChannel()会启动协程执行Reflector.Run()，同时接收到stopCh信号就会退出协程
56    wg.StartWithChannel(stopCh, r.Run)
57    // wait.Until()在前面的章节讲过了，周期性的调用c.processLoop()，这里来看是1秒
58    // 不用担心调用频率太高，正常情况下c.processLoop是不会返回的，除非遇到了解决不了的错误，因为他是个循环
59    wait.Until(c.processLoop, time.Second, stopCh)
60}
61
62// 代码源自client-go/tools/cache/controller.go
63func (c *controller) processLoop() {
64    for {
65        // 从队列中弹出一个对象，然后处理它,这才是最主要的部分，这个c.config.Process是构造Controller的时候通过Config传进来的
66        // 所以这个读者要特别注意了，这个函数其实是ShareInformer传进来的，所以在分析SharedInformer的时候要重点分析的
67        obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
68        if err != nil {
69            // 如果FIFO关闭了那就退出
70            if err == FIFOClosedError {
71                return
72            }
73            // 如果错误可以再试试
74            if c.config.RetryOnError {
75                c.config.Queue.AddIfNotPresent(obj)
76            }
77        }
78    }
79}
80
81// 代码源自client-go/tools/cache/controller.go
82// HasSynced() 调用的就是DeltaFIFO.HasSynced()实现的
83func (c *controller) HasSynced() bool {
84    return c.config.Queue.HasSynced()
85}
86// LastSyncResourceVersion() 是利用Reflector实现的
87func (c *controller) LastSyncResourceVersion() string {
88    if c.reflector == nil {
89        return ""
90    }
91    return c.reflector.LastSyncResourceVersion()
92}

  1// 代码源自client-go/informers/factory.go
  2// SharedInformerFactory是个interfaces，所以肯定有具体的实现类 
  3type SharedInformerFactory interface {
  4    // 在informers这个包中又定义了一个SharedInformerFactory，这个主要是包内抽象，所以此处继承了这个接口
  5    internalinterfaces.SharedInformerFactory
  6    // 这个暂时不知道干啥用，所以我也不介绍他了
  7    ForResource(resource schema.GroupVersionResource) (GenericInformer, error)
  8    // 等待所有的Informer都已经同步完成，这里同步其实就是遍历调用SharedInformer.HasSynced()
  9    // 所以函数需要周期性的调用指导所有的Informer都已经同步完毕
 10    WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
 11 
 12    Admissionregistration() admissionregistration.Interface // 返回admissionregistration相关的Informer组
 13    Apps() apps.Interface                                   // 返回app相关的Informer组
 14    Autoscaling() autoscaling.Interface                     // 返回autoscaling相关的Informer组
 15    Batch() batch.Interface                                 // 返回job相关的Informer组
 16    Certificates() certificates.Interface                   // 返回certificates相关的Informer组
 17    Coordination() coordination.Interface                   // 返回coordination相关的Informer组
 18    Core() core.Interface                                   // 返回core相关的Informer组
 19    Events() events.Interface                               // 返回event相关的Informer组
 20    Extensions() extensions.Interface                       // 返回extension相关的Informer组
 21    Networking() networking.Interface                       // 返回networking相关的Informer组
 22    Policy() policy.Interface                               // 返回policy相关的Informer组
 23    Rbac() rbac.Interface                                   // 返回rbac相关的Informer组
 24    Scheduling() scheduling.Interface                       // 返回scheduling相关的Informer组
 25    Settings() settings.Interface                           // 返回settings相关的Informer组
 26    Storage() storage.Interface                             // 返回storage相关的Informer组
 27}
 28
 29// 代码源自client-go/informers/internalinterfaces/factory_interfaces.go 
 30type SharedInformerFactory interface {
 31    // 核心逻辑函数，类似于很多类的Run()函数
 32    Start(stopCh <-chan struct{})
 33    // 这个很关键，通过对象类型，返回SharedIndexInformer，这个SharedIndexInformer管理的就是指定的对象
 34    // NewInformerFunc用于当SharedInformerFactory没有这个类型的Informer的时候创建使用
 35    InformerFor(obj runtime.Object, newFunc NewInformerFunc) cache.SharedIndexInformer
 36}
 37// 创建Informer的函数定义，这个函数需要apiserver的客户端以及同步周期，这个同步周期在SharedInformers反复提到
 38type NewInformerFunc func(kubernetes.Interface, time.Duration) cache.SharedIndexInformer
 39
 40
 41// 代码源自client-go/informers/factory.go
 42type sharedInformerFactory struct {
 43    // apiserver的客户端，暂时不用关心怎么实现的，只要知道他能列举和监听资源就可以了
 44    client           kubernetes.Interface
 45    // 哈哈，这样看来每个namesapce需要一个SharedInformerFactory，那cache用namespace建索引还有啥用呢？
 46    // 并不是所有的使用者都需要指定namesapce，比如kubectl，他就可以列举所有namespace的资源，所以他没有指定namesapce               
 47    namespace        string
 48    // 这是个函数指针，用来调整列举选项的，这个选项用来client列举对象使用
 49    tweakListOptions internalinterfaces.TweakListOptionsFunc
 50    // 互斥锁
 51    lock             sync.Mutex
 52    // 默认的同步周期，这个在SharedInformer需要用
 53    defaultResync    time.Duration
 54    // 每个类型的Informer有自己自定义的同步周期
 55    customResync     map[reflect.Type]time.Duration
 56    // 每类对象一个Informer，但凡使用SharedInformerFactory构建的Informer同一个类型其实都是同一个Informer
 57    informers map[reflect.Type]cache.SharedIndexInformer
 58    // 各种Informer启动的标记
 59    startedInformers map[reflect.Type]bool
 60}
 61// 代码源自client-go/tools/cache/shared_informer.go
 62// 这是一个通用的构造SharedInformerFactory的接口函数，没有任何其他的选项，只包含了apiserver的client以及同步周期
 63func NewSharedInformerFactory(client kubernetes.Interface, defaultResync time.Duration) SharedInformerFactory {
 64    // 最终是调用NewSharedInformerFactoryWithOptions()实现的，无非没有选项而已
 65    return NewSharedInformerFactoryWithOptions(client, defaultResync)
 66}
 67// 相比于上一个通用的构造函数，这个构造函数增加了namesapce过滤和调整列举选项
 68func NewFilteredSharedInformerFactory(client kubernetes.Interface, defaultResync time.Duration, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) SharedInformerFactory {
 69    // 最终是调用NewSharedInformerFactoryWithOptions()实现的，无非选项是2个
 70    // WithNamespace()和WithTweakListOptions()会在后文讲解
 71    return NewSharedInformerFactoryWithOptions(client, defaultResync, WithNamespace(namespace), WithTweakListOptions(tweakListOptions))
 72}
 73// 到了构造SharedInformerFactory核心函数了，其实SharedInformerOption是个有意思的东西
 74// 我们写程序喜欢Option是个结构体，但是这种方式的扩展很麻烦，这里面用的是回调函数，这个让我get到新技能了
 75func NewSharedInformerFactoryWithOptions(client kubernetes.Interface, defaultResync time.Duration, options ...SharedInformerOption) SharedInformerFactory {
 76    // 默认只有apiserver的client以及同步周期是需要外部提供的其他的都是可以有默认值的
 77    factory := &sharedInformerFactory{
 78        client:           client,
 79        namespace:        v1.NamespaceAll,
 80        defaultResync:    defaultResync,
 81        informers:        make(map[reflect.Type]cache.SharedIndexInformer),
 82        startedInformers: make(map[reflect.Type]bool),
 83        customResync:     make(map[reflect.Type]time.Duration),
 84    }
 85 
 86    //逐一遍历各个选项函数，opt是选项函数，下面面有详细介绍
 87    for _, opt := range options {
 88        factory = opt(factory)
 89    }
 90 
 91    return factory
 92}
 93// 代码源自client-go/informers/factory.go
 94// 这个是SharedInformerFactory构造函数的选项，是一个函数指针，传入的是工厂指针，返回也是工厂指针
 95// 很明显，选项函数直接修改工厂对象，然后把修改的对象返回就可以了
 96type SharedInformerOption func(*sharedInformerFactory) *sharedInformerFactory
 97// 把每个对象类型的同步周期这个参数转换为SharedInformerOption类型
 98func WithCustomResyncConfig(resyncConfig map[v1.Object]time.Duration) SharedInformerOption {
 99    // 这个实现很简单了，我就不多解释了
100    return func(factory *sharedInformerFactory) *sharedInformerFactory {
101        for k, v := range resyncConfig {
102            factory.customResync[reflect.TypeOf(k)] = v
103        }
104        return factory
105    }
106}
107// 这个选项函数用于使用者自定义client的列举选项的
108func WithTweakListOptions(tweakListOptions internalinterfaces.TweakListOptionsFunc) SharedInformerOption {
109    return func(factory *sharedInformerFactory) *sharedInformerFactory {
110        factory.tweakListOptions = tweakListOptions
111        return factory
112    }
113}
114// 这个选项函数用来设置namesapce过滤的
115func WithNamespace(namespace string) SharedInformerOption {
116    return func(factory *sharedInformerFactory) *sharedInformerFactory {
117        factory.namespace = namespace
118        return factory
119    }
120}
121
122// 代码源自client-go/informers/factory.go
123// 其实sharedInformerFactory的Start()函数就是启动所有具体类型的Informer的过程
124// 因为每个类型的Informer都是SharedIndexInformer，需要需要把每个SharedIndexInformer都要启动起来
125func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
126    // 加锁操作
127    f.lock.Lock()
128    defer f.lock.Unlock()
129    // 遍历informers这个map
130    for informerType, informer := range f.informers {
131        // 看看这个Informer是否已经启动过
132        if !f.startedInformers[informerType] {
133            // 如果没启动过，那就启动一个协程执行SharedIndexInformer的Run()函数，我们在分析SharedIndexInformer的时候
134            // 我们知道知道Run()是整个Informer的启动入口点，看了《深入浅出kubernetes之client-go的SharedInformer》
135            // 的同学应该会想Run()是谁调用的呢？这里面应该给你们答案了吧？
136            go informer.Run(stopCh)
137            // 设置Informer已经启动的标记
138            f.startedInformers[informerType] = true
139        }
140    }
141}
142
143// 代码源自client-go/informers/factory.go
144// InformerFor()相当于每个类型Informer的构造函数了，即便具体实现构造的地方是使用者提供的
145// 这个函数需要使用者传入对象类型，因为在sharedInformerFactory里面是按照对象类型组织的Informer
146// 更有趣的是这些Informer不是sharedInformerFactory创建的，需要使用者传入构造函数
147// 这样做既保证了每个类型的Informer只构造一次，同时又保证了具体Informer构造函数的私有化能力
148func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
149    // 加锁操作
150    f.lock.Lock()
151    defer f.lock.Unlock()
152    // 通过反射获取obj的类型
153    informerType := reflect.TypeOf(obj)
154    // 看看这个类型的Informer是否已经创建了？
155    informer, exists := f.informers[informerType]
156    // 如果Informer已经创建，那么就复用这个Informer
157    if exists {
158        return informer
159    }
160    // 获取这个类型定制的同步周期，如果定制的同步周期那就用统一的默认周期
161    resyncPeriod, exists := f.customResync[informerType]
162    if !exists {
163        resyncPeriod = f.defaultResync
164    }
165    // 调用使用者提供构造函数，然后把创建的Informer保存起来
166    informer = newFunc(f.client, resyncPeriod)
167    f.informers[informerType] = informer
168 
169    return informer
170}
171// 代码源自client-go/informers/internalinterfaces/factory_interfaces.go
172// 这个函数定义就是具体类型Informer的构造函数，后面会有地方说明如何使用
173type NewInformerFunc func(kubernetes.Interface, time.Duration) cache.SharedIndexInformer