NSQ源码分析(四)——inFlightPqueue和PriorityQueue优先级队列

本文主要是介绍NSQ源码分析(四)——inFlightPqueue和PriorityQueue优先级队列，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在Channel结构体中用到了两种优先级队列pqueue.PriorityQueue和inFlightPqueue。

deferredMessages map[MessageID]*pqueue.Item
deferredPQ       pqueue.PriorityQueue
deferredMutex    sync.MutexinFlightMessages map[MessageID]*Message
inFlightPQ       inFlightPqueue
inFlightMutex    sync.Mutex

其中deferredMessages和inFlightMessages 使用map储存了MessageID和Message的对应关系，用于根据MessageID获取对应的Message。而deferredPQ和inFlightPQ是两种优先级队列。deferredPQ队列中储存了延时消息和消息投递失败需要等待指定时间后重新投递的消息，inFlightPQ队列中储存了正在投递但还没确认投递成功的消息。

一、PriorityQueue优先级队列

源码位置在 nsq/internal/pqueue/pqueue.go文件中

PriorityQueue实现了Golang源码包heap中的接口，是最小堆。在了解PriorityQueue之前需要先对堆的概念有了解，可以参考：

https://blog.csdn.net/skh2015java/article/details/83183681

PriorityQueue队列中储存的是Item指针，Item结构体及字段说明如下：

type Item struct {Value interface{} //储存的消息内容Priority int64 //优先级的时间点Index int //在切片中的索引值}

这三个方法用于排序，按Priority字段排序

func (pq PriorityQueue) Len() int {return len(pq)}func (pq PriorityQueue) Less(i, j int) bool {return pq[i].Priority < pq[j].Priority}func (pq PriorityQueue) Swap(i, j int) {pq[i], pq[j] = pq[j], pq[i]pq[i].Index = ipq[j].Index = j}

常用方法：

New函数用于初始化队列并指定cap

func New(capacity int) PriorityQueue {return make(PriorityQueue, 0, capacity)}

Push函数用于向队列中添加元素

Pop函数用于取出队列中优先级最高的元素（即Priority值最小，也就是根部元素）

PeekAndShift(max int64) 用于判断根部的元素是否超过max，如果超过则返回nil，如果没有则返回并移除根部元素（根部元素是最小值）

在实际的项目中Priority字段存的是时间戳，比如说5分钟之后投递本条消息，则Priority字段存的就是5分钟之后的时间戳。而PeekAndShift(max int64)中max值就是当前时间，如果队列中的根部元素大于当前时间戳max的值，则说明队列中没有可以投递的消息，故返回nil。如果小于等于，则根部元素存放的消息可以投递，就是就返回并移除该根部元素。

func (pq *PriorityQueue) PeekAndShift(max int64) (*Item, int64) {if pq.Len() == 0 {return nil, 0}item := (*pq)[0] //获取根部元素if item.Priority > max {return nil, item.Priority - max}heap.Remove(pq, 0) //移除根部元素，并重新排列堆，使根部元素为最小值return item, 0}