如何解析client

共计 7833 个字符，预计需要花费 20 分钟才能阅读完成。

今天就跟大家聊聊有关如何解析 client-go 中 workqueue，可能很多人都不太了解，为了让大家更加了解，丸趣 TV 小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。

下面主要讲述下 client-go 中 workqueue, 看一下 client-go 的一个整体数据走向. 如下图：

而 workqueue 主要是在 listener 这里引用，listener 使用 chan 获取到数据之后将数据放入到工作队列进行处理。主要是由于 chan 过于简单，已经无法满足 K8S 的场景，所以衍生出了 workqueue，

特性

有序

去重

并发

延迟处理

限速

当前有三种 workqueue

基本队列

延迟队列

限速队列

其中延迟队列是基于基本队列实现的，而限流队列基于延迟队列实现

基本队列

看一下基本队列的接口

// 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

基本队列的 hello world 也很简单

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

基本队列 Add

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()}

基本队列 Get

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
 ...
}

主要是使用 chan 来保存延迟添加的元素, 而具体实现是通过一个实现了一个 AddAfter 方法，看一下具体的内容

// 延迟队列的接口
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:

上面 waitingLoop 是在实例化延迟队列的时候调用的，看一下实例化时候的逻辑

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
}

限速队列

当前限速队列支持 4 中限速模式

令牌桶算法限速

排队指数限速

计数器模式

混合模式 (多种限速算法同时使用)

限速队列的底层实际上还是通过延迟队列来进行限速, 通过计算出元素的限速时间作为延迟时间

来看一下限速接口

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))
}

令牌桶算法

client-go 中的令牌桶限速是通过 golang.org/x/time/rat 包来实现的

可以通过 flowcontrol.NewTokenBucketRateLimiter(qps float32, burst int) 来使用令牌桶限速算法，其中第一个参数 qps 表示每秒补充多少 token,burst 表示总 token 上限为多少。

排队指数算法

排队指数可以通过 workqueue.NewItemExponentialFailureRateLimiter(baseDelay time.Duration, maxDelay time.Duration) 来使用。

这个算法有两个参数:

baseDelay 基础限速时间

maxDelay 最大限速时间

举个例子来理解一下这个算法，例如快速插入 5 个相同元素，baseDelay 设置为 1 秒，maxDelay 设置为 10 秒，都在同一个限速期内。第一个元素会在 1 秒后加入到队列，第二个元素会在 2 秒后加入到队列，第三个元素会在 4 秒后加入到队列，第四个元素会在 8 秒后加入到队列，第五个元素会在 10 秒后加入到队列 (指数计算的结果为 16，但是最大值设置了 10 秒)。

来看一下源码的计算

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
}

计数器模式

计数器模式可以通过 workqueue.NewItemFastSlowRateLimiter(fastDelay, slowDelay time.Duration, maxFastAttempts int) 来使用，有三个参数

fastDelay 快限速时间

slowDelay 慢限速时间

maxFastAttempts 快限速元素个数

原理是这样的，假设 fastDelay 设置为 1 秒，slowDelay 设置为 10 秒，maxFastAttempts 设置为 3，同样在一个限速周期内快速插入 5 个相同的元素。前三个元素都是以 1 秒的限速时间加入到队列，添加第四个元素时开始使用 slowDelay 限速时间，也就是 10 秒后加入到队列，后面的元素都将以 10 秒的限速时间加入到队列，直到限速周期结束。

来看一下源码

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}
}

在 k8s-client-go 的源码中可以看到，大量的接口组合运用，将各种功能拆分成各个细小的库，是一种非常值得学习的代码风格以及思路。

看完上述内容，你们对如何解析 client-go 中 workqueue 有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注丸趣 TV 行业资讯频道，感谢大家的支持。