聊聊tempo的ExclusiveQueues
Posted codecraft
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了聊聊tempo的ExclusiveQueues相关的知识,希望对你有一定的参考价值。
序
本文主要研究一下tempo的ExclusiveQueues
ExclusiveQueues
tempo/pkg/flushqueues/exclusivequeues.go
type ExclusiveQueues struct {
queues []*util.PriorityQueue
index *atomic.Int32
activeKeys sync.Map
}
ExclusiveQueues定义了queues、index、activeKeys属性
New
tempo/pkg/flushqueues/exclusivequeues.go
// New creates a new set of flush queues with a prom gauge to track current depth
func New(queues int, metric prometheus.Gauge) *ExclusiveQueues {
f := &ExclusiveQueues{
queues: make([]*util.PriorityQueue, queues),
index: atomic.NewInt32(0),
}
for j := 0; j < queues; j++ {
f.queues[j] = util.NewPriorityQueue(metric)
}
return f
}
New方法先创建ExclusiveQueues,然后根据指定的queue个数通过util.NewPriorityQueue(metric)创建PriorityQueue
Enqueue
tempo/pkg/flushqueues/exclusivequeues.go
// Enqueue adds the op to the next queue and prevents any other items to be added with this key
func (f *ExclusiveQueues) Enqueue(op util.Op) {
_, ok := f.activeKeys.Load(op.Key())
if ok {
return
}
f.activeKeys.Store(op.Key(), struct{}{})
f.Requeue(op)
}
Enqueue方法先从activeKeys查找指定的key,若已经存在则提前返回,不存在则放入activeKeys中,然后执行f.Requeue(op)
Requeue
tempo/pkg/flushqueues/exclusivequeues.go
// Requeue adds an op that is presumed to already be covered by activeKeys
func (f *ExclusiveQueues) Requeue(op util.Op) {
flushQueueIndex := int(f.index.Inc()) % len(f.queues)
f.queues[flushQueueIndex].Enqueue(op)
}
Requeue方法首先通过int(f.index.Inc()) % len(f.queues)
计算flushQueueIndex,然后找到对应的queue,执行Enqueue方法
Dequeue
tempo/pkg/flushqueues/exclusivequeues.go
// Dequeue removes the next op from the requested queue. After dequeueing the calling
// process either needs to call ClearKey or Requeue
func (f *ExclusiveQueues) Dequeue(q int) util.Op {
return f.queues[q].Dequeue()
}
Dequeue方法执行f.queues[q]对应queue的Dequeue
Clear
tempo/pkg/flushqueues/exclusivequeues.go
// Clear unblocks the requested op. This should be called only after a flush has been successful
func (f *ExclusiveQueues) Clear(op util.Op) {
f.activeKeys.Delete(op.Key())
}
Clear方法将指定key从activeKeys中移除
Stop
tempo/pkg/flushqueues/exclusivequeues.go
// Stop closes all queues
func (f *ExclusiveQueues) Stop() {
for _, q := range f.queues {
q.Close()
}
}
Stop方法遍历f.queues,挨个执行q.Close()
小结
tempo的ExclusiveQueues定义了queues、index、activeKeys属性;它提供了Enqueue、Requeue、Dequeue、Clear、Stop方法。
doc
- tempo
以上是关于聊聊tempo的ExclusiveQueues的主要内容,如果未能解决你的问题,请参考以下文章
sql Canção - Tempo Consulta Item MRS