// Queue的Interface
type Interface interface {
// 添加对象
Add(item interface{})
// 队列长度
Len() int
// 获取对象，shutdown表示队列是否已经关闭了
Get() (item interface{}, shutdown bool)
// 告知队列对象已经处理完成，可以Pop了
Done(item interface{})
// 关闭队列
ShutDown()
// 队列是否正在关闭
ShuttingDown() bool
}
// Type is a work queue (see the package comment).
type Type struct {
// 队列存储数据结构，是一个数组
queue []t

// dirty这个集合包含了所有需要被操作的对象
dirty set

// 正在进行操作的对象的集合
processing set

// 信号量,条件同步
cond *sync.Cond

// 正在关闭标志
shuttingDown bool

// 监控用
metrics queueMetrics

// 更新没有完成的任务的周期，监控用
unfinishedWorkUpdatePeriod time.Duration
// 时钟
clock                      clock.Clock
}

// 空类型
type empty struct{}
// 伪范型2333
type t interface{}
// 集合的实现
type set map[t]empty

// 创建一个默认的队列
func New() *Type
// 创建一个有名字的队列
func NewNamed(name string) *Type
// 创建一个新的队列的实际执行函数
func newQueue(c clock.Clock, metrics queueMetrics, updatePeriod time.Duration) *Type
// 判断集合中是否含有对象t
func (s set) has(item t) bool
// 添加一个对象到集合
func (s set) insert(item t)
// 从集合中删除一个对象
func (s set) delete(item t)
// 添加一个对象到队列
// 先添加到dirty集合中
// 然后添加到queue中，
// 中间如果dirty/processing中已经存在，则返回
func (q *Type) Add(item interface{})
// 返回队列长度，即queue的长度
func (q *Type) Len() int
// pop queue的第一个元素，并添加到processing集合中
func (q *Type) Get() (item interface{}, shutdown bool)
// 通知某个对象被处理晚了，将对象从processing中移除，如果在dirty中
// 仍然存在该对象，说明该对象又一次变更了，所以需要再次添加到队列中
func (q *Type) Done(item interface{})
// 关闭队列
func (q *Type) ShutDown()
// 判断队列是否正在关闭
func (q *Type) ShuttingDown() bool
// 更新队列中没有完成的任务，监控使用
func (q *Type) updateUnfinishedWorkLoop()

// DelayingInterface是一种可以在添加一个对象前先等待一段时间的队列，这样
// 在对象处理失败，再次添加进队列的过程不会变成一个占用大量计算资源的无限循环
type DelayingInterface interface {
Interface
// 在等待duration之后，添加对象到队列
AddAfter(item interface{}, duration time.Duration)
}
// deplayingType包装了基本队列接口，并且提供了延时重新入队的实现
type delayingType struct {
Interface

// 时钟，用来确认延时执行的时机
clock clock.Clock

// 用于接收关闭的信号
stopCh chan struct{}
// 保证shutdown信号只被处理一次
stopOnce sync.Once

// 定时器，确保等待时间不超过maxWait
heartbeat clock.Ticker

// 等待添加元素的队列
waitingForAddCh chan *waitFor

// 重试次数的监控
metrics retryMetrics
}

// 保存对象和它应该被添加进队列的时机
type waitFor struct {
data    t
readyAt time.Time
// index in the priority queue (heap)
// 对象位于优先队列的索引
index int
}
// waitForPriorityQueue implements a priority queue for waitFor items.
//
// waitForPriorityQueue implements heap.Interface. The item occurring next in
// time (i.e., the item with the smallest readyAt) is at the root (index 0).
// Peek returns this minimum item at index 0. Pop returns the minimum item after
// it has been removed from the queue and placed at index Len()-1 by
// container/heap. Push adds an item at index Len(), and container/heap
// percolates it into the correct location.
// waitFor的优先队列，以readyAt的值小出队，实现了heap.Interface，从index 0出队，index Len()入队，
// 然后由heap操作保证正确顺序
type waitForPriorityQueue []*waitFor

// 创建一个匿名的延时队列
func NewDelayingQueue() DelayingInterface
// 创建一个有名字的延时队列
func NewNamedDelayingQueue(name string) DelayingInterface
// 创建一个延时队列，并启动waitingloop，创建延时队列的实际函数
func newDelayingQueue(clock clock.Clock, name string) DelayingInterface
// 返回优先队列长度
func (pq waitForPriorityQueue) Len() int
// 返回i和j的比较值，这里比较的是他们的readyAt值谁更小
func (pq waitForPriorityQueue) Less(i, j int) bool
// 交换i和j
func (pq waitForPriorityQueue) Swap(i, j int)
// 添加x到队尾
func (pq *waitForPriorityQueue) Push(x interface{})
// 从队列首抛出第一个对象
func (pq *waitForPriorityQueue) Pop() interface{}
// 获取队列首的对象，但是不从队列中移除
func (pq waitForPriorityQueue) Peek() interface{}
// 关闭队列
func (q *delayingType) ShutDown()
// 向waitingForAddCh添加一个waitFor对象
func (q *delayingType) AddAfter(item interface{}, duration time.Duration)
// waitingLoop在队列关闭之前，持续从队列中检查可以添加的对象
//  * 通过waitingEntryByData来避免重复添加，而改成更新readAt
func (q *delayingType) waitingLoop()
// 添加对象到队列，如果对象已经存在，就更新其readAt
func insert(q *waitForPriorityQueue, knownEntries map[t]*waitFor, entry *waitFor)

// RateLimitingInterface is an interface that rate limits items being added to the queue.
// 限速队列的Interface
type RateLimitingInterface interface {
DelayingInterface

// 以特定速度限制添加对象到队列，依赖特定的限速器进行判断
AddRateLimited(item interface{})

// 让限速器放弃重试某个元素。开发者对这个元素进行Done操作不受影响
Forget(item interface{})

// 返回某个对象被重新入队多少次，监控用
NumRequeues(item interface{}) int
}
// 限速队列的实现，由一个限速器和延时队列组成
type rateLimitingType struct {
DelayingInterface

rateLimiter RateLimiter
}

// 限速器接口，是限速队列的核心实现
type RateLimiter interface {
// 确认一个对象被添加前应该等待多久
When(item interface{}) time.Duration
// 放弃追踪一个对象
Forget(item interface{})
// 返回某个对象被重新入队多少次，监控用
NumRequeues(item interface{}) int
}

// 固定速率（qps)的限速器
type BucketRateLimiter struct {
*rate.Limiter
}

// 一种常用限速器，当对象处理失败的时候，其再次入队的等待时间 × 2，到MaxDelay为止
// 直到超过最大失败次数
type ItemExponentialFailureRateLimiter struct {
// 修改失败次数用到的锁
failuresLock sync.Mutex
failures     map[interface{}]int

baseDelay time.Duration
maxDelay  time.Duration
}

// 快慢限速器，对于一次对象，先以fastDelay为周期进行尝试，超过maxFastAttempts次数后，按照slowDelay为周期进行尝试
type ItemFastSlowRateLimiter struct {
failuresLock sync.Mutex
failures     map[interface{}]int

maxFastAttempts int
fastDelay       time.Duration
slowDelay       time.Duration
}

// MaxOfRateLimiter calls every RateLimiter and returns the worst case response
// When used with a token bucket limiter, the burst could be apparently exceeded in cases where particular items
// were separately delayed a longer time.
// 符合速率限制器，选择多个所有限速器中速度最慢的一种方案
type MaxOfRateLimiter struct {
limiters []RateLimiter
}

// 创建一个匿名的限速队列
func NewRateLimitingQueue(rateLimiter RateLimiter) RateLimitingInterface
// 创建一个有名字的限速队列
func NewNamedRateLimitingQueue(rateLimiter RateLimiter, name string) RateLimitingInterface
// AddRateLimited AddAfter's the item based on the time when the rate limiter says it's ok
// 添加一个元素到限速队列中，本质上是调用延时队列的接口，然后由限速器when接口来判断应该延迟多久
func (q *rateLimitingType) AddRateLimited(item interface{})
// 返回该对象的请求数量
func (q *rateLimitingType) NumRequeues(item interface{}) int
// 放弃追中一个对象
func (q *rateLimitingType) Forget(item interface{})

// 创建一个默认的限速器，由ItemExponentialFailureRateLimiter和
// BucketRateLimiter组成的复合限速器
func DefaultControllerRateLimiter() RateLimiter
// BucketRateLimiter的when实现
// 通过rate.Limiter.Reserve().Delay()获取延迟的周期
func (r *BucketRateLimiter) When(item interface{}) time.Duration
// 固定频率，不需要重试
func (r *BucketRateLimiter) NumRequeues(item interface{}) int
// 不需要重试，因此也不需要忘记
func (r *BucketRateLimiter) Forget(item interface{})
// 创建一个新的ItemExponentialFailureRateLimiter
func NewItemExponentialFailureRateLimiter(baseDelay time.Duration, maxDelay time.Duration) RateLimiter
// 默认的对象限速器，基础是每微秒重试1次，最长为1000s
func DefaultItemBasedRateLimiter() RateLimiter
// 根据失败次数，时间间隔将会不断翻倍，这里要注意判断是否超过MaxInt64造成溢出
func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration
// 返回重试次数，即failure[item]
func (r *ItemExponentialFailureRateLimiter) NumRequeues(item interface{}) int
// 放弃追踪对象，情况failure中的信息
func (r *ItemExponentialFailureRateLimiter) Forget(item interface{})
// 创建一个新的快慢限速器
func NewItemFastSlowRateLimiter(fastDelay, slowDelay time.Duration, maxFastAttempts int) RateLimiter
// 根据failture[item]中的情况，返回fastDelay或者slowDelay
func (r *ItemFastSlowRateLimiter) When(item interface{}) time.Duration
// 返回failture中的记录，即重试次数
func (r *ItemFastSlowRateLimiter) NumRequeues(item interface{}) int
// 放弃追踪一个对象，清除failure中的记录
func (r *ItemFastSlowRateLimiter) Forget(item interface{})
// 遍历所有的限速器，选择周期最长的那一个
func (r *MaxOfRateLimiter) When(item interface{}) time.Duration
// 创建一个新的限速器
func NewMaxOfRateLimiter(limiters ...RateLimiter) RateLimiter
// 返回所有限速器中重试次数最多的那个
func (r *MaxOfRateLimiter) NumRequeues(item interface{}) int
// 调用所有Limiter，放弃追踪一个对象
func (r *MaxOfRateLimiter) Forget(item interface{})

├── artifacts # crd定义和样例
├── code-of-conduct.md
├── CONTRIBUTING.md
├── controller.go # Controller定义
├── controller_test.go # 测试代码
├── docs # 文档
├── Godeps # 依赖
├── go.mod
├── go.sum
├── hack # 代码生成工具
├── LICENSE
├── main.go  # 程序入口，负责创建Controller
├── OWNERS
├── pkg # CRD的属性相关定义，中间涉及到客户端和deepcopy的部分可以用代码生成器生成
├── README.md
└── SECURITY_CONTACTS