本文基于Kubernetes v1.22.4版本进行源码学习，对应的client-go版本为v0.22.4

3、Informer机制

在Kubernetes系统中，组件之间通过HTTP协议进行通信，在不依赖任何中间件的情况下需要保证消息的实时性、可靠性、顺序性等。那么Kubernetes是如何做到的呢？答案就是Informer机制。Kubernetes的其他组件都是通过client-go的Informer机制与Kubernetes API Server进行通信的

1）、Informer架构

Informer架构设计中，有多个核心组件：

1. Reflector：用于监听Kubernetes资源，当资源发生变化时，触发相应的变更事件，例如Added、Updated、Deleted事件，并将其资源对象存放到本地缓存DeltaFIFO中
2. DeltaFIFO：可以分开理解，FIFO是一个先进先出的队列，它拥有队列操作的基本方法，例如Add、Update、Delete、List、Pop、Close等，而Delta是一个资源对象存储，它可以保存资源对象的操作类型，例如Added、Updated、Deleted、Sync等
3. Indexer：client-go中用来存储资源对象并自带索引功能的本地存储，Informer从DeltaFIFO中将消费出来的资源对象存储至Indexer。Indexer中的数据与etcd集群中的数据保持完全一致。client-go可以很方便地从本地存储中读取相应的资源对象数据，而无须每次都从远程etcd集群中读取，这样可以减轻Kubernetes API Server和etcd集群的压力

2）、Reflector

Reflector从Kubernetes API Server中listAndWatch资源对象，然后将对象的变化包装成Delta并放入到DeltaFIFO中

Reflector首先通过List操作获取全量的资源对象数据，调用DeltaFIFO的Replace方法全量插入DeltaFIFO，然后后续通过Watch操作根据资源对象的变化类型相应的调用DeltaFIFO的Add、Update、Delete方法，将对象及其变化插入到DeltaFIFO中

1）Reflector初始化

代码路径：vendor/k8s.io/client-go/tools/cache/reflector.go

func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
	return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod)
}

func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
	realClock := &clock.RealClock{}
	r := &Reflector{
		name:          name,
		listerWatcher: lw,
		store:         store,
		// We used to make the call every 1sec (1 QPS), the goal here is to achieve ~98% traffic reduction when
		// API server is not healthy. With these parameters, backoff will stop at [30,60) sec interval which is
		// 0.22 QPS. If we don't backoff for 2min, assume API server is healthy and we reset the backoff.
		backoffManager:         wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, realClock),
		initConnBackoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, realClock),
		resyncPeriod:           resyncPeriod,
		clock:                  realClock,
		watchErrorHandler:      WatchErrorHandler(DefaultWatchErrorHandler),
	}
	r.setExpectedType(expectedType)
	return r
}

通过NewReflector实例化Reflector对象时必须传入ListerWatcher interface的实现，它拥有List和Watch方法，用于获取及监控资源列表

2）ListWatch

代码路径：vendor/k8s.io/client-go/tools/cache/listwatch.go

type ListFunc func(options metav1.ListOptions) (runtime.Object, error)

type WatchFunc func(options metav1.ListOptions) (watch.Interface, error)

type ListWatch struct {
	ListFunc  ListFunc
	WatchFunc WatchFunc
	// DisableChunking requests no chunking for this list watcher.
	DisableChunking bool
}

ListWatch struct实现了ListerWatcher interface

再来看下ListWatch struct初始化的例子：

代码路径：vendor/k8s.io/client-go/informers/core/v1/pod.go

func NewPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
	return NewFilteredPodInformer(client, namespace, resyncPeriod, indexers, nil)
}

func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
	return cache.NewSharedIndexInformer(
		&cache.ListWatch{
			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.CoreV1().Pods(namespace).List(context.TODO(), options)
			},
			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.CoreV1().Pods(namespace).Watch(context.TODO(), options)
			},
		},
		&corev1.Pod{},
		resyncPeriod,
		indexers,
	)
}

在NewPodInformer初始化Pod对象的informer中，会初始化ListWatch struct并定义ListFunc和WatchFunc，可以看到ListFunc和WatchFunc即为其资源对象客户端的List与Watch方法

3）ListAndWatch函数

在Reflector源码实现中，其中最重要的是ListAndWatch函数，它负责获取资源列表和监听指定的Kubernetes API Server资源，ListAndWatch函数实现可分为三部分：List操作、Resync操作、Watch操作

a）List操作

List在程序第一次运行时获取该资源下所有的对象数据并将其存储至DeltaFIFO中，相关源码如下：

代码路径：vendor/k8s.io/client-go/tools/cache/reflector.go

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
	// 1.List操作(只执行一次)
	klog.V(3).Infof("Listing and watching %v from %s", r.expectedTypeName, r.name)
	var resourceVersion string

	// 1)设置ListOptions,将resourceVersion设置为"0"
	options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}

	if err := func() error {
		initTrace := trace.New("Reflector ListAndWatch", trace.Field{"name", r.name})
		defer initTrace.LogIfLong(10 * time.Second)
		var list runtime.Object
		var paginatedResult bool
		var err error
		listCh := make(chan struct{}, 1)
		panicCh := make(chan interface{}, 1)
		// 2)调用r.listerWatcher.List方法,执行list操作,获取全量的资源对象
		go func() {
			defer func() {
				if r := recover(); r != nil {
					panicCh <- r
				}
			}()
			// Attempt to gather list in chunks, if supported by listerWatcher, if not, the first
			// list request will return the full response.
			pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return r.listerWatcher.List(opts)
			}))
			switch {
			case r.WatchListPageSize != 0:
				pager.PageSize = r.WatchListPageSize
			case r.paginatedResult:
				// We got a paginated result initially. Assume this resource and server honor
				// paging requests (i.e. watch cache is probably disabled) and leave the default
				// pager size set.
			case options.ResourceVersion != "" && options.ResourceVersion != "0":
				// User didn't explicitly request pagination.
				//
				// With ResourceVersion != "", we have a possibility to list from watch cache,
				// but we do that (for ResourceVersion != "0") only if Limit is unset.
				// To avoid thundering herd on etcd (e.g. on master upgrades), we explicitly
				// switch off pagination to force listing from watch cache (if enabled).
				// With the existing semantic of RV (result is at least as fresh as provided RV),
				// this is correct and doesn't lead to going back in time.
				//
				// We also don't turn off pagination for ResourceVersion="0", since watch cache
				// is ignoring Limit in that case anyway, and if watch cache is not enabled
				// we don't introduce regression.
				pager.PageSize = 0
			}

			list, paginatedResult, err = pager.List(context.Background(), options)
			if isExpiredError(err) || isTooLargeResourceVersionError(err) {
				r.setIsLastSyncResourceVersionUnavailable(true)
				// Retry immediately if the resource version used to list is unavailable.
				// The pager already falls back to full list if paginated list calls fail due to an "Expired" error on
				// continuation pages, but the pager might not be enabled, the full list might fail because the
				// resource version it is listing at is expired or the cache may not yet be synced to the provided
				// resource version. So we need to fallback to resourceVersion="" in all to recover and ensure
				// the reflector makes forward progress.
				list, paginatedResult, err = pager.List(context.Background(), metav1.ListOptions{ResourceVersion: r.relistResourceVersion()})
			}
			close(listCh)
		}()
		select {
		case <-stopCh:
			return nil
		case r := <-panicCh:
			panic(r)
		case <-listCh:
		}
		if err != nil {
			return fmt.Errorf("failed to list %v: %v", r.expectedTypeName, err)
		}

		// We check if the list was paginated and if so set the paginatedResult based on that.
		// However, we want to do that only for the initial list (which is the only case
		// when we set ResourceVersion="0"). The reasoning behind it is that later, in some
		// situations we may force listing directly from etcd (by setting ResourceVersion="")
		// which will return paginated result, even if watch cache is enabled. However, in
		// that case, we still want to prefer sending requests to watch cache if possible.
		//
		// Paginated result returned for request with ResourceVersion="0" mean that watch
		// cache is disabled and there are a lot of objects of a given type. In such case,
		// there is no need to prefer listing from watch cache.
		if options.ResourceVersion == "0" && paginatedResult {
			r.paginatedResult = true
		}

		r.setIsLastSyncResourceVersionUnavailable(false) // list was successful
		initTrace.Step("Objects listed")
		listMetaInterface, err := meta.ListAccessor(list)
		if err != nil {
			return fmt.Errorf("unable to understand list result %#v: %v", list, err)
		}
		// 3)根据list返回的结果,获取最新的resourceVersion
		resourceVersion = listMetaInterface.GetResourceVersion()
		initTrace.Step("Resource version extracted")
		// 4)将list返回的结果转换为[]runtime.Object
		items, err := meta.ExtractList(list)
		if err != nil {
			return fmt.Errorf("unable to understand list result %#v (%v)", list, err)
		}
		initTrace.Step("Objects extracted")
		// 5)调用r.syncWith,将资源对象列表和resourceVersion存储(Replace)至DeltaFIFO中
		if err := r.syncWith(items, resourceVersion); err != nil {
			return fmt.Errorf("unable to sync list result: %v", err)
		}
		initTrace.Step("SyncWith done")
		// 6)调用r.setLastSyncResourceVersion,更新Reflector中已被处理的最新资源对象的resourceVersion值
		r.setLastSyncResourceVersion(resourceVersion)
		initTrace.Step("Resource version updated")
		return nil
	}(); err != nil {
		return err
	}
  ...

List操作（只执行一次）逻辑如下：

1. 设置ListOptions，将resourceVersion设置为"0"
2. 调用r.listerWatcher.List方法，执行list操作，获取全量的资源对象
3. 根据list返回的结果，获取最新的resourceVersion
4. 将list返回的结果转换为[]runtime.Object
5. 调用r.syncWith，将资源对象列表和resourceVersion存储（Replace）至DeltaFIFO中
6. 调用r.setLastSyncResourceVersion，更新Reflector中已被处理的最新资源对象的resourceVersion值

resourceVersion的作用：

• 保证客户端数据一致性和顺序性
• 乐观锁，实现并发控制

设置ListOptions时，resourceVersion有三种设置方法：

• 不设置，此时会直接从etcd中读取，此时数据是最新的
• 设置为"0"，此时会从API Server Cache中获取数据
• 设置为指定的resourceVersion，获取resourceVersion大于指定版本的所有资源对象

b）Resync操作

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
  ...
	// 2.Resync操作(异步循环执行)
	resyncerrc := make(chan error, 1)
	cancelCh := make(chan struct{})
	defer close(cancelCh)
	go func() {
		resyncCh, cleanup := r.resyncChan()
		defer func() {
			cleanup() // Call the last one written into cleanup
		}()
		for {
			select {
			case <-resyncCh:
			case <-stopCh:
				return
			case <-cancelCh:
				return
			}
			// 1)判断是否需要执行Resync操作,即重新同步
			if r.ShouldResync == nil || r.ShouldResync() {
				klog.V(4).Infof("%s: forcing resync", r.name)
				// 2)如果需要,则调用r.store.Resync进行重新同步
				if err := r.store.Resync(); err != nil {
					resyncerrc <- err
					return
				}
			}
			cleanup()
			resyncCh, cleanup = r.resyncChan()
		}
	}()
  ...

Resync操作（异步循环执行）逻辑如下：

1. 判断是否需要执行Resync操作，即重新同步
2. 如果需要，则调用r.store.Resync进行重新同步

c）Watch操作

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
	...
	// 3.Watch操作(循环执行)
	for {
		// 1)根据stopCh判断是否需要退出循环
		// give the stopCh a chance to stop the loop, even in case of continue statements further down on errors
		select {
		case <-stopCh:
			return nil
		default:
		}

		timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
		// 2)设置ListOptions,将resourceVersion设置为最新的resourceVersion,即从list返回的最新的resourceVersion开始执行watch操作
		options = metav1.ListOptions{
			ResourceVersion: resourceVersion,
			// We want to avoid situations of hanging watchers. Stop any wachers that do not
			// receive any events within the timeout window.
			TimeoutSeconds: &timeoutSeconds,
			// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
			// Reflector doesn't assume bookmarks are returned at all (if the server do not support
			// watch bookmarks, it will ignore this field).
			AllowWatchBookmarks: true,
		}

		// start the clock before sending the request, since some proxies won't flush headers until after the first watch event is sent
		start := r.clock.Now()
		// 3)调用r.listerWatcher.Watch,开始监听操作
		w, err := r.listerWatcher.Watch(options)
		if err != nil {
			// If this is "connection refused" error, it means that most likely apiserver is not responsive.
			// It doesn't make sense to re-list all objects because most likely we will be able to restart
			// watch where we ended.
			// If that's the case begin exponentially backing off and resend watch request.
			// Do the same for "429" errors.
			if utilnet.IsConnectionRefused(err) || apierrors.IsTooManyRequests(err) {
				<-r.initConnBackoffManager.Backoff().C()
				continue
			}
			return err
		}

		// 4)调用r.watchHandler,处理watch操作返回来的结果
		if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {
			if err != errorStopRequested {
				switch {
				case isExpiredError(err):
					// Don't set LastSyncResourceVersionUnavailable - LIST call with ResourceVersion=RV already
					// has a semantic that it returns data at least as fresh as provided RV.
					// So first try to LIST with setting RV to resource version of last observed object.
					klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.expectedTypeName, err)
				case apierrors.IsTooManyRequests(err):
					klog.V(2).Infof("%s: watch of %v returned 429 - backing off", r.name, r.expectedTypeName)
					<-r.initConnBackoffManager.Backoff().C()
					continue
				default:
					klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err)
				}
			}
			return nil
		}
	}
}  

Watch操作（循环执行）逻辑如下：

1. 根据stopCh判断是否需要退出循环
2. 设置ListOptions，将resourceVersion设置为最新的resourceVersion，即从list返回的最新的resourceVersion开始执行watch操作
3. 调用r.listerWatcher.Watch，开始监听操作
4. 调用r.watchHandler，处理watch操作返回来的结果

watchHandler用于处理资源的变更事件。当触发Added、Updated、Deleted事件时，将对应的资源对象更新到本地缓存DeltaFIFO并更新resourceVersion。代码如下：

func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
	eventCount := 0

	// Stopping the watcher should be idempotent and if we return from this function there's no way
	// we're coming back in with the same watch interface.
	defer w.Stop()

loop:
	for {
		select {
		case <-stopCh:
			return errorStopRequested
		case err := <-errc:
			return err
		// 1)从watch操作返回来的结果中获取event事件
		case event, ok := <-w.ResultChan():
			if !ok {
				break loop
			}
			if event.Type == watch.Error {
				return apierrors.FromObject(event.Object)
			}
			if r.expectedType != nil {
				if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
					utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
					continue
				}
			}
			if r.expectedGVK != nil {
				if e, a := *r.expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a {
					utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", r.name, e, a))
					continue
				}
			}
			meta, err := meta.Accessor(event.Object)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
				continue
			}
			// 2)获得当前watch到资源的resourceVersion
			newResourceVersion := meta.GetResourceVersion()
			// 3)当触发Added、Updated、Deleted事件时,将对应的资源对象更新到本地缓存DeltaFIFO
			switch event.Type {
			case watch.Added:
				err := r.store.Add(event.Object)
				if err != nil {
					utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err))
				}
			case watch.Modified:
				err := r.store.Update(event.Object)
				if err != nil {
					utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", r.name, event.Object, err))
				}
			case watch.Deleted:
				// TODO: Will any consumers need access to the "last known
				// state", which is passed in event.Object? If so, may need
				// to change this.
				err := r.store.Delete(event.Object)
				if err != nil {
					utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err))
				}
			case watch.Bookmark:
				// A `Bookmark` means watch has synced here, just update the resourceVersion
			default:
				utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
			}
			*resourceVersion = newResourceVersion
			// 4)调用r.setLastSyncResourceVersion,更新Reflector中已被处理的最新资源对象的resourceVersion值
			r.setLastSyncResourceVersion(newResourceVersion)
			if rvu, ok := r.store.(ResourceVersionUpdater); ok {
				rvu.UpdateResourceVersion(newResourceVersion)
			}
			eventCount++
		}
	}

	watchDuration := r.clock.Since(start)
	if watchDuration < 1*time.Second && eventCount == 0 {
		return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
	}
	klog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedTypeName, eventCount)
	return nil
}

3）、DeltaFIFO

DeltaFIFO可以分开理解，FIFO是一个先进先出的队列，它拥有队列操作的基本方法，例如Add、Update、Delete、List、Pop、Close等，而Delta是一个资源对象存储，它可以保存资源对象的操作类型，例如Added、Updated、Deleted、Sync等

1）DeltaFIFO结构体

代码路径：vendor/k8s.io/client-go/tools/cache/delta_fifo.go

type DeltaFIFO struct {
	...
	// 存放Delta,与queue中存放的key是同样的key
	items map[string]Deltas

	// 存储资源对象的key,可以确保顺序性
	queue []string

	// 默认使用MetaNamespaceKeyFunc,默认使用<namespace>/<name>的格式,不指定namespace时用<name>
	keyFunc KeyFunc
	...
}

type Deltas []Delta

type Delta struct {
	Type   DeltaType
	Object interface{}
}

type DeltaType string

const (
	Added   DeltaType = "Added"
	Updated DeltaType = "Updated"
	Deleted DeltaType = "Deleted"
	Replaced DeltaType = "Replaced"
	Sync DeltaType = "Sync"
)

DeltaFIFO会保留所有关于资源对象的操作类型，队列中会存在拥有不同操作类型的同一资源对象，消费者在处理该资源对象时能够了解该资源对象所发生的事情

queue字段存储资源对象的key，该key通过keyFunc计算得出，默认使用MetaNamespaceKeyFunc，默认使用<namespace>/<name>的格式，不指定namespace时用<name>

items字段通过map数据结构的方式存储，value存储的是对象的Delta数组

DeltaFIFO存储结构如下图所示：

DeltaFIFO本质上是一个先进先出的队列，有数据的生产者和消费者：

生产过程：

• Reflector的List
• Reflector的Watch
• Reflector的Resync

消费过程：

• 事件派发到WorkQueue
• 刷新本地缓存

2）生产者方法

DeltaFIFO队列中的资源对象在Added、Updated、Deleted事件中都调用了queueActionLocked函数，它是DeltaFIFO实现的关键，代码如下：

func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
	// 1)计算出资源对象的key
	id, err := f.KeyOf(obj)
	if err != nil {
		return KeyError{obj, err}
	}
	oldDeltas := f.items[id]
	// 2)将actionType和资源对象构造成Delta,添加到items中,并通过dedupDeltas函数进行去重操作
	newDeltas := append(oldDeltas, Delta{actionType, obj})
	newDeltas = dedupDeltas(newDeltas)

	if len(newDeltas) > 0 {
		if _, exists := f.items[id]; !exists {
			f.queue = append(f.queue, id)
		}
		f.items[id] = newDeltas
		// 3)更新构造后的Delta并通过cond.Broadcast通知所有消费者解除阻塞
		f.cond.Broadcast()
	} else {
		// This never happens, because dedupDeltas never returns an empty list
		// when given a non-empty list (as it is here).
		// If somehow it happens anyway, deal with it but complain.
		if oldDeltas == nil {
			klog.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; ignoring", id, oldDeltas, obj)
			return nil
		}
		klog.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; breaking invariant by storing empty Deltas", id, oldDeltas, obj)
		f.items[id] = newDeltas
		return fmt.Errorf("Impossible dedupDeltas for id=%q: oldDeltas=%#+v, obj=%#+v; broke DeltaFIFO invariant by storing empty Deltas", id, oldDeltas, obj)
	}
	return nil
}

queueActionLocked逻辑如下：

1. 通过f.KeyOf函数计算出资源对象的key
2. 将actionType和资源对象构造成Delta，添加到items中，并通过dedupDeltas函数进行去重操作
3. 更新构造后的Delta并通过cond.Broadcast通知所有消费者解除阻塞

3）消费者方法

Pop方法作为消费者方法使用，从DeltaFIFO的头部取出最早进入队列中的资源对象数据。Pop方法需要传入process函数，用于接收并处理对象的回调方法，代码如下：

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
	f.lock.Lock()
	defer f.lock.Unlock()
	for {
		for len(f.queue) == 0 {
			// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
			// When Close() is called, the f.closed is set and the condition is broadcasted.
			// Which causes this loop to continue and return from the Pop().
			if f.closed {
				return nil, ErrFIFOClosed
			}
			// 1)当队列中没有数据时,通过f.cond.Wait阻塞等待数据,只有收到cond.Broadcast时才说明有数据被添加,解决当前阻塞状态
			f.cond.Wait()
		}
		// 2)如果队列中不为空,取出f.queue的头部数据,将该对象传入process回调函数,由上层消费者进行处理
		id := f.queue[0]
		f.queue = f.queue[1:]
		if f.initialPopulationCount > 0 {
			f.initialPopulationCount--
		}
		item, ok := f.items[id]
		if !ok {
			// This should never happen
			klog.Errorf("Inconceivable! %q was in f.queue but not f.items; ignoring.", id)
			continue
		}
		delete(f.items, id)
		err := process(item)
		// 3)如果process回调函数处理错误,则将该对象重新存入队列
		if e, ok := err.(ErrRequeue); ok {
			f.addIfNotPresent(id, item)
			err = e.Err
		}
		// Don't need to copyDeltas here, because we're transferring
		// ownership to the caller.
		return item, err
	}
}

pop逻辑如下：

1. 当队列中没有数据时，通过f.cond.Wait阻塞等待数据，只有收到cond.Broadcast时才说明有数据被添加,解决当前阻塞状态
2. 如果队列中不为空，取出f.queue的头部数据，将该对象传入process回调函数，由上层消费者进行处理
3. 如果process回调函数处理错误，则将该对象重新存入队列

参考：

《Kubernetes源码剖析》

2022年最新k8s编程operator篇

k8s client-go源码分析 informer源码分析(3)-Reflector源码分析