workqueue

​ k8s中的各个controller控制器内都有使用workqueue做事件队列，为了满足各个控制器不同队列需求，k8s client-go内置各种符合需求的workqueue队列，workqueue不仅实现了队列的基本功能，还实现了去重、延迟、限频等功能。

1.简介

• 去重：当一个资源对象被频繁变更，同一个对象还未被消费时，没必要在队列中存多份，经过去重后只需处理一次。
• 延迟：部分控制器需要延迟队列功能，类似cronjob依赖延迟队列实现定时功能，另外也可以实现延迟backoff时长后重入队。
• 限频：避免过多事件并发入队，使用限频策略对入队的事件个数进行控制，k8s中的controller大把的使用限频。
• 
• workqueue中内置了三种队列模型，interface会实现基本的先进先出队列，与常规队列相比多了去重功能，DelayingInterface在interface的基础上实现了延迟队列的功能，RateLimitingInterface又在DelayingInterface基础上实现了RateLimiter限频器功能，当插入元素的次数超过限频器规则时，把对象推到延迟队列中处理。

2.原理分析

2.1.queue

2.1.1.接口定义

• queue实现基本的先进先出队列，跟常规队列相比多了去重功能，它的接口定义了基本队列该有的方法。

• type Interface interface {
      // 添加元素
  	Add(item interface{})
      // 获取队列的长度，queue字段的长度
  	Len() int
      // 获取队列元素
  	Get() (item interface{}, shutdown bool)
      // 标记元素执行完毕
  	Done(item interface{})
      // 关闭
  	ShutDown()
      // 优雅关闭
  	ShutDownWithDrain()
  	// 正在关闭
      ShuttingDown() bool
  }
  

2.1.2.queue结构

• type Type struct {
  	// 切片存放未处理的元素
  	queue []t
  
  	// 去重队列，避免未消费队列存放相同元素
  	dirty set
  
  	// 正在处理队列，避免相同元素并发执行
  	processing set
  
      // 条件变量，用于唤醒等待元素的协程
  	cond *sync.Cond
  	
  	// 指标统计
  	metrics queueMetrics
  }
  

• queue用来实现顺序存储元素，其结构为slice切片类型，元素类型为interface{}任意类型，queue的读写流程为读slice的头部，写slice的尾部，queue是FIFO先进先出的设计。

• dirty set用来实现去重，所以它使用了set数据结构，但是它的去重只是针对待消费的元素。

• processing set也是用来去重的，其主要规避元素被并发处理。当元素还未被处理时，通过dirty去重，当前queue只有一个元素；当元素已经被执行，但还未调用done标记完成，这个时候同一个元素再入队，会放到dirty去重实现排队效果。

• 

2.1.3.Add元素

• Add()为元素插入到队列的方法，插入元素的流程原理包括：

  1.判断dirty是否存在该元素，存在则跳过处理，实现待处理元素的去重效果

  2.添加元素到dirty，判断processing集合是否存在该元素，存在则跳过，实现元素并发处理控制

  3.添加元素到queue待处理队列，唤醒其他阻塞协程

• func (q *Type) Add(item interface{}) {
      // 加锁保证并发安全
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
  	if q.shuttingDown {
  		return
  	}
      // dirty已经存在直接退出
  	if q.dirty.has(item) {
  		return
  	}
  
      // 增加add指标
  	q.metrics.add(item)
  
      // add元素到dirty
  	q.dirty.insert(item)
      // 如果某个元素正在处理则直接结束，此时该元素已经放在dirty，由Done方法处理dirty-->queue逻辑
  	if q.processing.has(item) {
  		return
  	}
  
      // 元素放入待处理队列
  	q.queue = append(q.queue, item)
      
      // 唤醒其他协程
  	q.cond.Signal()
  }
  

2.1.4.Get元素

• Get用于获取元素，从队列的头部获取最先入队的元素然后在processing队列添加元素，其目的是为了防止同一个元素被并发处理，最后从dirty队列删除元素，因为dirty是为了实现待消费去重，从queue中拿走元素，dirty中也需要删除。

• func (q *Type) Get() (item interface{}, shutdown bool) {
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
      // queue队列为空则执行wait阻塞等待
  	for len(q.queue) == 0 && !q.shuttingDown {
  		q.cond.Wait()
  	}
      // 如果workqueue关闭且queue为空，结束处理
  	if len(q.queue) == 0 {
  		// We must be shutting down.
  		return nil, true
  	}
  
      // 从头部获取元素
  	item = q.queue[0]
  	// The underlying array still exists and reference this object, so the object will not be garbage collected.
  	q.queue[0] = nil
      // 重新引用切片
  	q.queue = q.queue[1:]
  
      // 统计metric的get指标
  	q.metrics.get(item)
  
      // 从dirty元素中去除，加入到processing集合
  	q.processing.insert(item)
  	q.dirty.delete(item)
  
  	return item, false
  }
  

2.1.5.Done元素

• Done()用来标记某元素是否处理完，然后判断dirty中是否存在该对象，存在则将该对象推入queue里再次入队，通过processing标记该元素正在被处理但还未完成，其目的就是为了防止一个元素被并发同时处理。这意味着一个元素正在被处理，如果再次添加同一个元素，由于该元素还在处理未完成，只能把对象放在dirty里（queue slice的元素并发场景会被多个协程处理），当执行完毕调用Done()时，会把dirty的任务重新入队，起到排队效果。

• func (q *Type) Done(item interface{}) {
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
  
      // 统计metric done指标
  	q.metrics.done(item)
  
      // 从processing集合中删除该元素
  	q.processing.delete(item)
      // 处理dirty入queue逻辑
  	if q.dirty.has(item) {
  		q.queue = append(q.queue, item)
  		q.cond.Signal()
  	} else if q.processing.len() == 0 {
  		q.cond.Signal()
  	}
  }
  

2.2.delayingqueue

2.2.1.接口定义

• delayingqueue是在queue interface基础上实现的延迟队列，这意味着它可以使用queue interface的接口方法，并进一步扩充了AddAfter方法。

• type DelayingInterface interface {
  	Interface
  	// 添加定时功能
  	AddAfter(item interface{}, duration time.Duration)
  }
  
  type delayingType struct {
      // 继承queue interface的基本功能
  	Interface
  
  	// clock tracks time for delayed firing
  	clock clock.Clock
  
  	// 退出管道
  	stopCh chan struct{}
  	stopOnce sync.Once
  
  	// 周期性检测队列中对象是否到期
  	heartbeat clock.Ticker
  
  	// 新的定时元素会推到该管道，等待loop处理
  	waitingForAddCh chan *waitFor
  
  	// metrics指标统计
  	metrics retryMetrics
  }
  

2.2.2.初始化模块

• delayingqueue首先会使用小顶堆minheap排列定时任务，当添加定时任务时，把该任务扔到一个chan里，然后由一个独立的协程监听该chan，把任务扔到heap里，该独立协程会从堆里找到最近的到期任务，对该任务进行到期监听。当定时到期后，会把到期的任务推入queue队列。

• // 心跳的时长
  const maxWait = 10 * time.Second
  
  // 构建定时器队列对象
  func NewDelayingQueueWithCustomClock(clock clock.WithTicker, name string) DelayingInterface {
      // clock为k8s内部封装的时间对象，NewNamed用于生成queue
  	return newDelayingQueue(clock, NewNamed(name), name)
  }
  
  func newDelayingQueue(clock clock.WithTicker, q Interface, name string) *delayingType {
  	ret := &delayingType{
  		Interface:       q,
  		clock:           clock,
  		heartbeat:       clock.NewTicker(maxWait),
  		stopCh:          make(chan struct{}),
  		waitingForAddCh: make(chan *waitFor, 1000),
  		metrics:         newRetryMetrics(name),
  	}
  
      // 初始化堆并监听延迟对象
  	go ret.waitingLoop()
  	return ret
  }
  

2.2.3.延迟监听模块

• func (q *delayingType) waitingLoop() {
  	defer utilruntime.HandleCrash()
  
  	// Make a placeholder channel to use when there are no items in our list
  	never := make(<-chan time.Time)
  
  	// 计时器变量，用于记录队列中下一个任务的到期时间
  	var nextReadyAtTimer clock.Timer
  
      // 初始化minheap小顶堆
  	waitingForQueue := &waitForPriorityQueue{}
  	heap.Init(waitingForQueue)
  
      // 存储任务数据和等待对象的映射关系
  	waitingEntryByData := map[t]*waitFor{}
  
  	for {
          // queue关闭时退出
  		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
  			}
  
              // 到期则弹出对象放入queue
  			entry = heap.Pop(waitingForQueue).(*waitFor)
  			q.Add(entry.data)
  			delete(waitingEntryByData, entry.data)
  		}
  
  		// 准备等待下一个任务的时间
  		nextReadyAt := never
  		if waitingForQueue.Len() > 0 {
              // 如果之前有定时器运行，先停止
  			if nextReadyAtTimer != nil {
  				nextReadyAtTimer.Stop()
  			}
              // 从堆顶获取最近到期的元素
  			entry := waitingForQueue.Peek().(*waitFor)
              // 实例化timer定时器
  			nextReadyAtTimer = q.clock.NewTimer(entry.readyAt.Sub(now))
              // 将nextReadyAt通道设置为定时器的触发通道
  			nextReadyAt = nextReadyAtTimer.C()
  		}
  
  		select {
  		case <-q.stopCh:
  			return
  
          // 心跳信号定期触发，用于让循环继续运行，防止任务被遗漏
  		case <-q.heartbeat.C():
  			// 触发10s心跳超时后重新选择最近任务
  
  		case <-nextReadyAt:
              // 上次计算的最近元素的定时器已到期触发(nextReadyAtTimer触发的)，进行下次循环，期间会处理该到期任务
              // 就算nextReadyAt=nerver阻塞了这个case，其他case还是会触发，保证循环的继续进行
  
          // 收到新添加的定时器
  		case waitEntry := <-q.waitingForAddCh:
              // 如果新对象还未到期，则把定时对象放到heap定时堆中
  			if waitEntry.readyAt.After(q.clock.Now()) {
  				insert(waitingForQueue, waitingEntryByData, waitEntry)
  			} else {
                  // 如果新对象已经到期，放到queue中
  				q.Add(waitEntry.data)
  			}
  
              // 取尽标识，尽量在本地循环把chan读空，避免留存后select阶段总被唤醒
  			drained := false
  			for !drained {
  				select {
  				case waitEntry := <-q.waitingForAddCh:
  					if waitEntry.readyAt.After(q.clock.Now()) {
                          // 对象入堆并调整位置
  						insert(waitingForQueue, waitingEntryByData, waitEntry)
  					} else {
  						q.Add(waitEntry.data)
  					}
  				default:
  					drained = true
  				}
  			}
  		}
  	}
  }
  

2.2.4.添加定时任务

• // 调用方使用 AddAfter 添加定时任务
  func (q *delayingType) AddAfter(item interface{}, duration time.Duration) {
  	// don't add if we're already shutting down
  	if q.ShuttingDown() {
  		return
  	}
  
      // 进行统计
  	q.metrics.retry()
  
  	// 时间不合理，直接入队
  	if duration <= 0 {
  		q.Add(item)
  		return
  	}
  
  	select {
  	case <-q.stopCh:
  		// 等待退出
  	case q.waitingForAddCh <- &waitFor{data: item, readyAt: q.clock.Now().Add(duration)}:
          // 创建一个定时对象，  然后推到 waitingForAddCh 管道中，等待waitingLoop处理
  	}
  }
  

2.2.5.小顶堆实现

• type waitFor struct {
      // 存放元素
  	data    t
      // 时间点
  	readyAt time.Time
      
  	// 同一个时间点下对比递增的索引
  	index int
  }
  
  type waitForPriorityQueue []*waitFor
  
  func (pq waitForPriorityQueue) Len() int {
  	return len(pq)
  }
  
  func (pq waitForPriorityQueue) Less(i, j int) bool {
  	return pq[i].readyAt.Before(pq[j].readyAt)
  }
  
  func (pq waitForPriorityQueue) Swap(i, j int) {
  	pq[i], pq[j] = pq[j], pq[i]
  	pq[i].index = i
  	pq[j].index = j
  }
  
  func (pq *waitForPriorityQueue) Push(x interface{}) {
  	n := len(*pq)
  	item := x.(*waitFor)
  	item.index = n
  	*pq = append(*pq, item)
  }
  
  func (pq *waitForPriorityQueue) Pop() interface{} {
  	n := len(*pq)
  	item := (*pq)[n-1]
  	item.index = -1
  	*pq = (*pq)[0:(n - 1)]
  	return item
  }
  
  func (pq waitForPriorityQueue) Peek() interface{} {
  	return pq[0]
  }
  

2.2.5.原理总结

• delayingqueue通过监听nextReadyAt可以等待最近任务的到期，监听waitingForAddCh可以获取新增的时间，heartbeat定时器则是确保select分支至少每个一段时间执行一次而不会阻塞。另外delayingqueue复用了queue的底层逻辑，它在此基础上扩充了queue对于定时任务的需求。

• 

2.3.ratelimitingqueue

2.3.1.接口定义

• rateLimitingInterface是在delayingInterface 基础上实现的队列，k8s的controller等组件都有使用rateLimitingInterface。通过AddRateLimitd入队时，需要先经过rateLimiter计算是否触发限频，如需限频则计算该元素的delay时长，把该对象推到delayingInterface延迟队列处理。delayingInterface内部的waitingLoop协程会监听由AddRateLimited推入的定时任务，如果是延迟任务则放在heap里，否则立马推入queue。

• type RateLimitingInterface interface {
      // 继承了 DelayingInterface 延迟队列
      DelayingInterface
  
      // 使用对应的限频算法求出需要 dalay 的时常, 然后添加到 delay 队列中.
      AddRateLimited(item interface{})
  
      // 在 rateLimiter 中取消某对象的追踪记录.
      Forget(item interface{})
  
      // 从 rateLimiter 中获取计数.
      NumRequeues(item interface{}) int
  }
  
  type rateLimitingType struct {
      // 继承延迟队列的工鞥呢
      DelayingInterface
  
      // 限速组件
      rateLimiter RateLimiter
  }
  

2.3.2.代码实现

• 实例化rateLimitingType时，需要创建delayingInterface和配置rateLimiter。AddLimited方法通过限频器计算出需要等待的时长，然后调用delayingQueue.AddAfter()方法决定把对象仍在延迟队列还是普通队列。

  k8s控制器在使用限频类的workqueue时，当入队超过一定阈值后会采用异步的方法添加任务，这样对于k8s controller来说避免了同步等待。另外Forget()方法是在reateLimiter里清理掉某对象的相关记录，该接口涉及的rateLimiter需要视情况而定，并不是所有的rateLimiter都真正实现该接口。

• // 众多实例化方法之一, 传入 ratelimiter 限频器
  func NewRateLimitingQueue(rateLimiter RateLimiter) RateLimitingInterface {
      return &rateLimitingType{
          // 实例化延迟队列
          DelayingInterface: NewDelayingQueue(),
  
          // 限频器
          rateLimiter:       rateLimiter,
      }
  }
  
  // 实现了 RateLimitingInterface 接口
  type rateLimitingType struct {
      DelayingInterface
  
      rateLimiter RateLimiter
  }
  
  // 通过限频器计算出需要当代的时间, 如需要等待, 然后把对象扔到延迟队里.
  func (q *rateLimitingType) AddRateLimited(item interface{}) {
      q.DelayingInterface.AddAfter(item, q.rateLimiter.When(item))
  }
  
  // 从限频器里获取该对象的计数信息.
  func (q *rateLimitingType) NumRequeues(item interface{}) int {
      return q.rateLimiter.NumRequeues(item)
  }
  
  // 从限频器里删除该对象的记录的信息.
  func (q *rateLimitingType) Forget(item interface{}) {
      q.rateLimiter.Forget(item)
  }
  

2.4.3.rateLimiter

• type RateLimiter interface {
  	// When gets an item and gets to decide how long that item should wait
  	When(item interface{}) time.Duration
  	// Forget indicates that an item is finished being retried.  Doesn't matter whether it's for 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
  }
  

• workqueue内置了几个rateLimiter限频器，同时也支持自定义限频器，只需实现rateLimiter接口即可，内置的限频器包括：

  1.bucketRateLimiter，通过rate.Limiter限速

  2.ItemExponentialFailureRateLimiter，通过backoff进行限速

  3.ItemFastSlowRateLimiter，超过阈值使用fastDelay，否则使用slowDelay使用间隔

  4.ItemFastSlowRateLimiter，抽象了rateLimiter方法，可以同时对多个rateLimiter实例进行计算，最后求出合理值

2.4.4.令牌桶限速

• type BucketRateLimiter struct {
      *rate.Limiter
  }
  
  // 该类实现了 RateLimiter 接口
  var _ RateLimiter = &BucketRateLimiter{}
  
  func (r *BucketRateLimiter) When(item interface{}) time.Duration {
      // 通过 rate 获取新元素需要等待的时间.
      return r.Limiter.Reserve().Delay()
  }
  
  func (r *BucketRateLimiter) NumRequeues(item interface{}) int {
      // 直接返回 0.
      return 0
  }
  
  func (r *BucketRateLimiter) Forget(item interface{}) {
      // 暂未实现该方法.
  }
  

2.4.5.基于backoff限速

• type ItemExponentialFailureRateLimiter struct {
      failuresLock sync.Mutex
      failures     map[interface{}]int
  
      baseDelay time.Duration
      maxDelay  time.Duration
  }
  
  func NewItemExponentialFailureRateLimiter(baseDelay time.Duration, maxDelay time.Duration) RateLimiter {
      return &ItemExponentialFailureRateLimiter{
          failures:  map[interface{}]int{},
          baseDelay: baseDelay,
          maxDelay:  maxDelay,
      }
  }
  
  func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration {
      r.failuresLock.Lock()
      defer r.failuresLock.Unlock()
  
      // 获取上次计数, 且递增增加一.
      exp := r.failures[item]
      r.failures[item] = r.failures[item] + 1
  
      // 通过公式计算 backoff 时长, 当前时长为上次的二次方.
      backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2, float64(exp))
      if backoff > math.MaxInt64 {
          // 不能超过 maxDelay
          return r.maxDelay
      }
  
      // 把纳秒的时间戳转成 time duration
      calculated := time.Duration(backoff)
      if calculated > r.maxDelay {
          // 不能超过 maxDelay
          return r.maxDelay
      }
  
      return calculated
  }
  
  // 获取该对象的入队的次数.
  func (r *ItemExponentialFailureRateLimiter) NumRequeues(item interface{}) int {
      r.failuresLock.Lock()
      defer r.failuresLock.Unlock()
  
      return r.failures[item]
  }
  
  // 不在追踪该对象, 在这里是不记录该对象的次数.
  func (r *ItemExponentialFailureRateLimiter) Forget(item interface{}) {
      r.failuresLock.Lock()
      defer r.failuresLock.Unlock()
  
      delete(r.failures, item)
  }
  

• 使用map记录各个元素计数，后通过经典的backoff算法可以求出当前需要等待的时长，默认为1，只要不forget抹掉计数，那么再次入队时等待的时长为上次的二次方

2.4.6.maxOfRateLimiter

• type MaxOfRateLimiter struct {
      // 多个 ratelimiter 实例
      limiters []RateLimiter
  }
  
  func (r *MaxOfRateLimiter) When(item interface{}) time.Duration {
      ret := time.Duration(0)
      // 依次调用, 求最大的时长
      for _, limiter := range r.limiters {
          curr := limiter.When(item)
          if curr > ret {
              ret = curr
          }
      }
  
      return ret
  }
  
  // 创建入口
  func NewMaxOfRateLimiter(limiters ...RateLimiter) RateLimiter {
      return &MaxOfRateLimiter{limiters: limiters}
  }
  
  func (r *MaxOfRateLimiter) NumRequeues(item interface{}) int {
      ret := 0
      // 依次调用, 求最大
      for _, limiter := range r.limiters {
          curr := limiter.NumRequeues(item)
          if curr > ret {
              ret = curr
          }
      }
  
      return ret
  }
  
  func (r *MaxOfRateLimiter) Forget(item interface{}) {
      // 依次调用
      for _, limiter := range r.limiters {
          limiter.Forget(item)
      }
  }
  

• maxOfRateLimiter实例化时可以传入多个rateLimiter限速器实例，使用when求等待间隔时，遍历计算所有的rateLimiter实例，求最大的时长。Forget()同理，需要对所有的rateLimiter集合遍历调用。

2.4.分析

• client-go workqueue实现了三种队列类型，interface为最基本的队列类型，delayingInterface在interface的基础上实现了延迟队列，rateLimiterInterface又在delayingInterface基础上实现了限频队列。

  workqueue工作时，通过k8s informer监听资源的变更，实例化informer时需注册addFunc/updateFunc/deleteFunc事件方法。这些方法对应的操作就是把delta对象放到workqueue，控制器通常会开启多个协程进行队列消费，拿到的对象使用控制器的sync进行状态同步。

•