WorkQueue
WorkQueue称为工作队列,Kubernetes的WorkQueue队列与普通的FIFO对比,实现略显复杂,它的主要功能在于标记和去重,并支持以下特性:
有序:按照添加顺序处理元素(item)
去重:相同元素在同一时间不会被重复处理,例如一个元素在处理前被添加了多次,它只会被处理一次。
并发性:多生产者和多消费者
标记机制:支持标记功能,标记一个元素是否被处理,也允许元素在处理时重新排队。
通知机制:ShutDown方法通过信号量通知队列不再接收新的元素,并通知metric goroutine退出。
延迟:支持延迟队列,延迟一段时间后再将元素存入队列。
限速:支持限速队列,元素存入队列时进行速率限制。限制一个元素被重新排队(Reenqueued)的次数。
Metric:支持metric监控指标,可用于Prometheus监控。
WorkQueue支持3种队列,并提供3种接口,不同队列实现可应对不同的使用场景,分别如下:
Interface:FIFO队列接口,先进先出队列,并支持去重机制。
DelayingInterface:延迟队列接口,基于Interface接口封装,延迟一段时间后再将元素存入队列。
RateLimitingInterface:限速队列接口,基于DelayingInterface接口封装,支持元素存入队列时进行速率限制。
1. FIFO队列
FIFO队列是最基本的队列方法,例如增删改查,获取长度等。WorkQueue中的延迟队列和限速队列都是基于Interface接口实现的,Interface定义如下:
源码路径:k8s.io\client-go\util\workqueue\queue.go
type Interface interface {
//给队列添加元素(item),可以是任意类型
Add(item interface{})
//获取队列的长度
Len() int
//获取队列头部元素
Get() (item interface{}, shutdown bool)
//标记队列中该元素已被处理
Done(item interface{})
//关闭队列
ShutDown()
//查询队列是否正在关闭
ShuttingDown() bool
}
FIFO队列的数据结构为:
/ Type is a work queue (see the package comment).
type Type struct {
// queue defines the order in which we will work on items. Every
// element of queue should be in the dirty set and not in the
// processing set.
//用来存储元素的地方,slice结构,保证元素有序
queue []t
// dirty defines all of the items that need to be processed.
//dirty很重要,除了能够去重,还能保证在处理一个元素之前哪怕其被添加多次(并发),但也只会被处理一次
dirty set
// Things that are currently being processed are in the processing set.
// These things may be simultaneously in the dirty set. When we finish
// processing something and remove it from this set, we'll check if
// it's in the dirty set, and if so, add it to the queue.
//用于标记机制,标记一个元素是否正在被处理
processing set
cond *sync.Cond
shuttingDown bool
metrics queueMetrics
unfinishedWorkUpdatePeriod time.Duration
clock clock.Clock
}
// set是由map构造的
type empty struct{}
type t interface{}
type set map[t]empty
例如上图所示为FIFO的存储过程,通过Add方法向FIFO队列中分别插入1,2,3这3个元素,此时队列中的queue和dirty字段分别存有1,2,3元素,processing字段为空。
然后通过Get方法获取最先进入的元素(元素1),此时队列中的queue和dirty字段分别存有2,3;元素1被放入processing字段中,说明它正在被处理。最后处理完元素1时,通过Done方法将其标记为处理完成,此时队列中的processing字段中的1元素被删除。
但是再并发存储下,如何保证处理一个元素之前哪怕被添加多次,也只是处理一次,下图为FIFO并发存储的过程。
在并发场景下,goroutine A通过Get方法获取元素1,元素1被添加到processing字段中,同一时间,goroutine B通过Add方法插入另一个1元素,此时在processing字段中已经存在相同的元素,所以后面后面的元素1不会被直接插入到queue字段中,而是存入dirty字段中;在goroutine A通过Done方法标记处理完元素1后,如果dirty字段中存有元素1,则将其追加到queue字段的尾部,dirty和processing字段都是HashMap数据结构实现的,不考虑无序,只考虑去重。
下面来看源码,
源码路径:k8s.io\client-go\util\workqueue\queue.go
// Get方法
// Get blocks until it can return an item to be processed. If shutdown = true,
// the caller should end their goroutine. You must call Done with item when you
// have finished processing it.
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()
}
if len(q.queue) == 0 {
// We must be shutting down.
return nil, true
}
//获取第一个元素queue首个元素,并从queue中移除
item, q.queue = q.queue[0], q.queue[1:]
q.metrics.get(item)
//将插入到processing字段中
q.processing.insert(item)
//将item从dirty字段中移除
q.dirty.delete(item)
return item, false
}
//==========================================================================================
//Add方法
// Add marks item as needing processing.
func (q *Type) Add(item interface{}) {
q.cond.L.Lock()
defer q.cond.L.Unlock()
if q.shuttingDown {
return
}
//如果dirty中含有item,则返回,防止重复写入
if q.dirty.has(item) {
return
}
q.metrics.add(item)
//将item写入到dirty字段中
q.dirty.insert(item)
//如果processing字段中存在该元素,则直接返回
if q.processing.has(item) {
return
}
//否则,才将其插入到queue尾部
q.queue = append(q.queue, item)
//唤醒其他线程
q.cond.Signal()
}
//==========================================================================================
//Done方法
// Done marks item as done processing, and if it has been marked as dirty again
// while it was being processed, it will be re-added to the queue for
// re-processing.
func (q *Type) Done(item interface{}) {
q.cond.L.Lock()
defer q.cond.L.Unlock()
q.metrics.done(item)
//将处理完的item从processing中移除
q.processing.delete(item)
//判断dirty字段中是否含有item,如果有则将其添加到queue尾部
if q.dirty.has(item) {
q.queue = append(q.queue, item)
q.cond.Signal()
}
}
2. 延迟队列
延迟队列是基于FIFO队列接口封装的,在原有功能上增加了AddAfter方法,其原理是延迟一段时间后再将元素插入FIFO队列,延迟队列的数据结构如下:
// DelayingInterface is an Interface that can Add an item at a later time. This makes it easier to
// requeue items after failures without ending up in a hot-loop.
type DelayingInterface interface {
Interface
// AddAfter adds an item to the workqueue after the indicated duration has passed
//AddAfter方法插入一个item,并附带duration参数,该参数用于指定元素延迟插入FIFO队列的时间,如果小于0,则直接插入
AddAfter(item interface{}, duration time.Duration)
}
// delayingType wraps an Interface and provides delayed re-enquing
type delayingType struct {
Interface
// clock tracks time for delayed firing
clock clock.Clock
// stopCh lets us signal a shutdown to the waiting loop
stopCh chan struct{}
// stopOnce guarantees we only signal shutdown a single time
stopOnce sync.Once
// heartbeat ensures we wait no more than maxWait before firing
heartbeat clock.Ticker
// waitingForAddCh is a buffered channel that feeds waitingForAdd
//主要字段,默认初始大小为1000;当插入的元素大于或等于1000时,延迟队列才会处于阻塞状态,
//该字段的数据通过goroutine运行的waitingLoop函数持久运行
waitingForAddCh chan *waitFor
// metrics counts the number of retries
metrics retryMetrics
}
延迟队列运行原理如图所示:
将元素1放入waitingForAddCh字段中,通过waitingLoop函数消费元素数据,当元素的延迟时间不大于当前时间时,说明还需要延迟将元素插入FIFO队列的时间,此时将该元素放入优先队列(waitForPriorityQueue)中。当元素的延迟时间大于当前时间时,则将该元素插入FIFO队列中。同时,也会遍历waitForPriorityQueue中的元素,按照上述逻辑验证时间。
看下源码:
源码路径:k8s.io\client-go\util\workqueue\delaying_queue.go
// AddAfter adds the given item to the work queue after the given delay
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()
// immediately add things with no delay
//如果延迟时间小于等于0,则直接插入到queue中
if duration <= 0 {
q.Add(item)
return
}
select {
case <-q.stopCh:
// unblock if ShutDown() is called
// 在当前时间增加duration时间,构造waitFor类型放入q.waitingForAddCh中
case q.waitingForAddCh <- &waitFor{data: item, readyAt: q.clock.Now().Add(duration)}:
}
}
// 延迟队列初始化函数
func newDelayingQueue(clock clock.Clock, 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),
}
//goroutine运行waitingLoop函数
go ret.waitingLoop()
return ret
}
// =================================================================================================
//看一下waitingLoop函数核心代码
// waitingLoop runs until the workqueue is shutdown and keeps a check on the list of items to be added.
func (q *delayingType) waitingLoop() {
......
//初始化waitForPriorityQueue数组
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) {
//如果entry的延迟时间晚于now,说明还需延迟等待
break
}
//否则,将其从waitingForQueue中移出并插入到队列q中,同时删除map waitingEntryByData中的该元素的数据
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)
//从上面循环跳出后,获取waitingForQueue中第一个元素,判断其时间
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():
// continue the loop, which will add ready items
case <-nextReadyAt:
// continue the loop, which will add ready items
case waitEntry := <-q.waitingForAddCh:
//q.waitingForAddCh有数据输出时,获取该数据,并判断该数据是否需要延迟
if waitEntry.readyAt.After(q.clock.Now()) {
//若需要延迟,则插入到waitingForQueue,及waitingEntryByData
insert(waitingForQueue, waitingEntryByData, waitEntry)
} else {
//否则,插入到队列q中
q.Add(waitEntry.data)
}
// 不知用意,是否全部耗尽?否则循环等待?
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
}
}
}
}
}
3. 限速队列
限速对列是基于延迟队列和FIFO队列接口封装,限速队列接口(RateLimitingInterface)在原有功能上增加了AddRateLimited、Forget、NumRequeues方法。限速队列的重点不在于RateLimitingInterface接口,而在于它提供的四种限速算法接口,其原理是:限速队列利用延迟队列的特性,延迟某个元素的插入时间,达到限速的目的。数据结构为:
源码路径:k8s.io\client-go\util\workqueue\default_rate_limiters.go
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 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
}
When:获取指定元素应该等待的时间
Forget:释放指定元素,清空该元素的排队数
NumRequeues:获取指定元素的排队数
注意:限速周期——很重要的一个概念。
一个限速周期是指从执行AddRateLimited方法到执行完Forget方法之间的时间。如果该元素被Forget方法处理完,则清空排队数。
3.1 令牌桶算法(BucketRateLimiter)
令牌桶算法是通过Go的第三方库golang.org/x/time/rate实现。内部实现了一个存放token(令牌)的”桶“,初始时”桶“是空的,token会以固定速率往”桶“里填充,直到将其填满为止,多余的token会被丢弃。每个元素都会从令牌桶得到一个token,直到得到token的元素才允许通过(accept),而没有得到token的元素处于等待状态。令牌桶算法通过控制token发放来达到限速的目的。原理图如下:
WorkQueue在默认情况下会实例化令牌桶,代码如下:
&BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
在实例化rate.NewLimiter后,传入两个参数r和b,其中r参数表示每秒往”桶“里填充的token的数量,b参数表示令牌桶的大小(即令牌桶最多存放token的数量)。默认r=10,b=100;假设在一个限速周期内插入1000个元素,通过r.Limiter.Reserve().Delay函数返回指定元素应该等待的时间,那么前b(即100)个元素会被立刻处理,而后面元素的延迟时间分别为item100/100ms、item101/200ms、item102/300ms、item103/400ms,以此类推。
3.2 排队指数算法(ItemExponentialFailureLimiter)
排队指数算法将相同元素的排队数作为指数,排队数增大,速率限制呈指数级增长,但其最大值不会超过maxDelay。元素的排队数统计是有限速周期的。一个限速周期是从执行AddRateLimited方法到执行完Forget方法之间的时间。如果该元素被Forget方法处理完,则清空排队数,核心代码实现为:
源码路径:k8s.io\client-go\util\workqueue\default_rate_limiters.go
// ItemExponentialFailureRateLimiter does a simple baseDelay*2^<num-failures> limit
// dealing with max failures and expiration are up to the caller
type ItemExponentialFailureRateLimiter struct {
// 排队队列锁
failuresLock sync.Mutex
// 统计元素的排队数,每当AddRateLimited方法插入新元素时,会使该字段加1
failures map[interface{}]int
// 最初的限速单位,默认为5ms
baseDelay time.Duration
// 最大限速单位,默认为1000s
maxDelay time.Duration
}
func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration {
r.failuresLock.Lock()
defer r.failuresLock.Unlock()
//获取item元素排队数量并将其加1
exp := r.failures[item]
r.failures[item] = r.failures[item] + 1
// The backoff is capped such that 'calculated' value never overflows.
//计算其限速时间,指数运算
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
}
在同一限速周期内,如果不存在相同元素,那么所有元素的延迟时间为baseDelay;而在同一限速周期内,如果存在相同元素,那么相同元素的延迟时间呈指数级增长,最长时间不会超过maxDelay。
3.3 计数器算法(ItemFastSlowRateLimiter)
计算器算法是限速算法中最简单的一种,其原理是:限制一段时间内允许通过的元素数量,例如在1分钟内只允许通过100个元素,每插入一个元素,计数器自增1,当计数器到100的阈值且还在限速周期内时,则不允许元素再通过。WorkQueue在此基础上扩展了fast和slow速率。数据结构为:
// ItemFastSlowRateLimiter does a quick retry for a certain number of attempts, then a slow retry after that
type ItemFastSlowRateLimiter struct {
failuresLock sync.Mutex
//统计元素的排队数
failures map[interface{}]int
//控制从fast速率转换到slow速率
maxFastAttempts int
//fast速率
fastDelay time.Duration
//slow速率
slowDelay time.Duration
}
func (r *ItemFastSlowRateLimiter) When(item interface{}) time.Duration {
r.failuresLock.Lock()
defer r.failuresLock.Unlock()
//排队数加1
r.failures[item] = r.failures[item] + 1
//判断排队数是否超过限制fast速率的最大值,如果超过则返回slow速率
if r.failures[item] <= r.maxFastAttempts {
return r.fastDelay
}
return r.slowDelay
}
通过源码看到,当排队数超过maxFastAttempts字段限制的数量时,则返回slow速率,即假如maxFastAttempts=3,当第4个元素到达时,返回slow速率;则前三个元素使用fast速率,而第四个使用slow速率。
3.4 混合模式(MaxOfRateLimiter)
混合模式是将多种限速算法混合使用,即多种限速算法同时生效。下面为默认使用排队指数算法和令牌桶算法:
// DefaultControllerRateLimiter is a no-arg constructor for a default rate limiter for a workqueue. It has
// both overall and per-item rate limiting. The overall is a token bucket and the per-item is exponential
func DefaultControllerRateLimiter() RateLimiter {
return NewMaxOfRateLimiter(
NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second),
// 10 qps, 100 bucket size. This is only for retry speed and its only the overall factor (not per item)
&BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
)
}
Last updated
Was this helpful?