k8s-client-go源码剖析(三)

// client-go源码路径util/workqueue/queue.go
type Interface interface {
	//新增元素 可以是任意对象
	Add(item interface{})
	//获取当前队列的长度
	Len() int
	// 阻塞获取头部元素(先入先出)  返回元素以及队列是否关闭
	Get() (item interface{}, shutdown bool)
	// 显示标记完成元素的处理
	Done(item interface{})
	//关闭队列
	ShutDown()
	//队列是否处于关闭状态
	ShuttingDown() bool
}

type Type struct {
	//含有所有元素的元素的队列 保证有序
	queue []t
	//所有需要处理的元素 set是基于map以value为空struct实现的结构，保证去重
	dirty set
	//当前正在处理中的元素
	processing set
	...
}

type empty struct{}
type t interface{}
type set map[t]empty

 wq := workqueue.New()
	wq.Add("hello")
	v, _ := wq.Get()

func (q *Type) Add(item interface{}) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	//如果当前处于关闭状态,则不再新增元素
	if q.shuttingDown {
		return
	}
	//如果元素已经在等待处理中,则不再新增
	if q.dirty.has(item) {
		return
	}
	//添加到metrics
	q.metrics.add(item)
	//加入等待处理中
	q.dirty.insert(item)
	//如果目前正在处理该元素 就不将元素添加到队列
	if q.processing.has(item) {
		return
	}
	q.queue = append(q.queue, item)
	q.cond.Signal()
}

func (q *Type) Get() (item interface{}, shutdown bool) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	//如果当前没有元素并且不处于关闭状态,则阻塞
	for len(q.queue) == 0 && !q.shuttingDown {
		q.cond.Wait()
	}
	...
	item, q.queue = q.queue[0], q.queue[1:]
	q.metrics.get(item)
	//把元素添加到正在处理队列中
	q.processing.insert(item)
	//把队列从等待处理队列中删除
	q.dirty.delete(item)
	return item, false
}

func newQueue(c clock.Clock, metrics queueMetrics, updatePeriod time.Duration) *Type {
	t := &Type{
		clock:                      c,
		dirty:                      set{},
		processing:                 set{},
		cond:                       sync.NewCond(&sync.Mutex{}),
		metrics:                    metrics,
		unfinishedWorkUpdatePeriod: updatePeriod,
	}
        //启动一个协程 定时更新metrics
	go t.updateUnfinishedWorkLoop()
	return t
}

func (q *Type) updateUnfinishedWorkLoop() {
	t := q.clock.NewTicker(q.unfinishedWorkUpdatePeriod)
	defer t.Stop()
	for range t.C() {
		if !func() bool {
			q.cond.L.Lock()
			defer q.cond.L.Unlock()
			if !q.shuttingDown {
				q.metrics.updateUnfinishedWork()
				return true
			}
			return false

		}() {
			return
		}
	}
}

type delayingType struct {
	Interface
        ...
	//放置延迟添加的元素
	waitingForAddCh chan *waitFor
       ...
}

//延迟队列的接口
type DelayingInterface interface {
	Interface
	// AddAfter adds an item to the workqueue after the indicated duration has passed
	AddAfter(item interface{}, duration time.Duration)
}

func (q *delayingType) AddAfter(item interface{}, duration time.Duration) {
	...
	//如果延迟实现小于等于0 直接添加到队列
	if duration <= 0 {
		q.Add(item)
		return
	}
	select {
	case <-q.stopCh:
	//添加到chan,下面会讲一下这个chan的处理
	case q.waitingForAddCh <- &waitFor{data: item, readyAt: q.clock.Now().Add(duration)}:
	}
}

func (q *delayingType) waitingLoop() {

	defer utilruntime.HandleCrash()

	never := make(<-chan time.Time)

	var nextReadyAtTimer clock.Timer

	waitingForQueue := &waitForPriorityQueue{}
	//这里是初始化一个优先队列 具体实现有兴趣的同学可以研究下
	heap.Init(waitingForQueue)

	waitingEntryByData := map[t]*waitFor{}

	for {
		if q.Interface.ShuttingDown() {
			return
		}

		now := q.clock.Now()

		// Add ready entries
		for waitingForQueue.Len() > 0 {
			entry := waitingForQueue.Peek().(*waitFor)
			//看一下第一个元素是否已经到达延迟的时间了
			if entry.readyAt.After(now) {
				break
			}

			//时间到了,将元素添加到工作的队列,并且从延迟的元素中移除
			entry = heap.Pop(waitingForQueue).(*waitFor)
			q.Add(entry.data)
			delete(waitingEntryByData, entry.data)
		}

		// Set up a wait for the first item's readyAt (if one exists)
		nextReadyAt := never
		if waitingForQueue.Len() > 0 {
			if nextReadyAtTimer != nil {
				nextReadyAtTimer.Stop()
			}
			//如果还有需要延迟的元素,计算第一个元素的延迟时间(最小延迟的元素)
			entry := waitingForQueue.Peek().(*waitFor)
			nextReadyAtTimer = q.clock.NewTimer(entry.readyAt.Sub(now))
			nextReadyAt = nextReadyAtTimer.C()
		}

		select {
		case <-q.stopCh:
			return
		case <-q.heartbeat.C():
			//定时检查下是否有元素达到延迟的时间
		case <-nextReadyAt:
			//这里是上面计算出来的时间,时间到了,处理到达延迟时间的元素
		case waitEntry := <-q.waitingForAddCh:
			//检查是否需要延迟,如果需要延迟就加入到延迟等待
			if waitEntry.readyAt.After(q.clock.Now()) {
				insert(waitingForQueue, waitingEntryByData, waitEntry)
			} else {
				//如果不需要延迟就直接添加到队列
				q.Add(waitEntry.data)
			}

			drained := false
			for !drained {
				select {
				case waitEntry := <-q.waitingForAddCh: 

func NewDelayingQueueWithCustomClock(clock clock.Clock, name string) DelayingInterface {
	//实例化一个数据结构
	ret := &delayingType{
		Interface:       NewNamed(name),
		clock:           clock,
		heartbeat:       clock.NewTicker(maxWait),
		stopCh:          make(chan struct{}),
		waitingForAddCh: make(chan *waitFor, 1000),
		metrics:         newRetryMetrics(name),
	}

	//放到一个协程中处理延迟元素
	go ret.waitingLoop()

	return ret
}

type RateLimiter interface {
	//
	When(item interface{}) time.Duration
	// Forget indicates that an item is finished being retried.  Doesn't matter whether its for perm failing
	// or for success, we'll stop tracking it
	Forget(item interface{})
	// NumRequeues returns back how many failures the item has had
	NumRequeues(item interface{}) int
}

// RateLimitingInterface is an interface that rate limits items being added to the queue.
type RateLimitingInterface interface {
	DelayingInterface

	//实际上底层还是调用的延迟队列,通过计算出元素的延迟时间 进行限速
	AddRateLimited(item interface{})

	// Forget indicates that an item is finished being retried.  Doesn't matter whether it's for perm failing
	// or for success, we'll stop the rate limiter from tracking it.  This only clears the `rateLimiter`, you
	// still have to call `Done` on the queue.
	Forget(item interface{})

	// NumRequeues returns back how many times the item was requeued
	NumRequeues(item interface{}) int
}

func (q *rateLimitingType) AddRateLimited(item interface{}) {
         //通过when方法计算延迟加入队列的时间
	q.DelayingInterface.AddAfter(item, q.rateLimiter.When(item))
}

func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration {
	r.failuresLock.Lock()
	defer r.failuresLock.Unlock()

	//第一次为0
	exp := r.failures[item]
	//累加1
	r.failures[item] = r.failures[item] + 1

	//通过当前计数和baseDelay计算指数结果  baseDelay*(2的exp次方)
	backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2, float64(exp))
	if backoff > math.MaxInt64 {
		return r.maxDelay
	}

	calculated := time.Duration(backoff)
	if calculated > r.maxDelay {
		return r.maxDelay
	}

	return calculated
}

func (r *ItemFastSlowRateLimiter) When(item interface{}) time.Duration {
	r.failuresLock.Lock()
	defer r.failuresLock.Unlock()
	//添加一次就计数一次
	r.failures[item] = r.failures[item] + 1
	//计数小于maxFastAttempts都以fastDelay为限速时间，否则以slowDelay为限速时间
	if r.failures[item] <= r.maxFastAttempts {
		return r.fastDelay
	}
	return r.slowDelay
}

func NewMaxOfRateLimiter(limiters ...RateLimiter) RateLimiter {
	return &MaxOfRateLimiter{limiters: limiters}
}