Go实战 | 一文带你搞懂从单队列到优先级队列的实现

yudotyang · · 679 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

原文链接:https://mp.weixin.qq.com/s/bPLRcsmSO5_MvqN8F812zQ 大家好,我是「Go学堂」的渔夫子,今天跟大家聊聊在我们项目中的优先级队列的实现及应用场景。 ### 优先级队列概述 队列,是数据结构中实现先进先出策略的一种数据结构。而优先队列则是带有优先级的队列,即先按优先级分类,然后相同优先级的再 进行排队。优先级高的队列中的元素会优先被消费。如下图所示: ![图1-优先级队列概况图.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/28e6b7aaf6544871944b8ff62f874c4c~tplv-k3u1fbpfcp-watermark.image?) 在Go中,可以定义一个切片,切片的每个元素代表一种优先级队列,切片的索引顺序代表优先级顺序,后面代码实现部分我们会详细讲解。 ### 为什么需要优先级队列 先来看现实生活中的例子。银行的办事窗口,有普通窗口和vip窗口,vip窗口因为排队人数少,等待的时间就短,比普通窗口就会优先处理。同样,在登机口,就有贵宾通道和普通,同样贵宾通道优先登机。 在互联网中,当然就是请求和响应。使用优先级队列的作用是**将请求按特定的属性划分出优先级,然后按优先级的高低进行优先处理**。在研发服务的时候这里有个隐含的约束条件就是**服务器资源(CPU、内存、带宽等)是有限的**。如果服务器资源是无限的,那么也就不需要队列进行排队了,来一个请求就立即处理一个请求就好了。所以,为了在最大限度的利用服务器资源的前提下,将更重要的任务(优先级高的请求)优先处理,以更好的服务用户。 对于请求优先级的划分可以根据业务的特点根据**价值高的优先原则**来进行划分即可。例如可以根据是否是否是会员、是否是VIP会员等属性进行划分优先级。也可以根据是否是付费用户进行划分。在博客的业务中,也可以根据是否是大V的属性进行优先级划分。在互联网广告业务中,可以根据广告位资源价值高低来划分优先级。 ### 优先级队列实现原理 #### 01 四个角色 在完整的优先级队列中有四个角色,分别是**优先级队列、工作单元、消费者worker、通知channel**。 - **工作单元Job**:队列里的元素。我们把每一次业务处理都封装成一个工作单元,该工作单元会进入对应的优先级队列进行排队,然后等待消费者worker来消费执行。 - **优先级队列**:按优先级划分的队列,用来暂存对应优先级的工作单元Job,相同优先级的工作单元会在同一个队列里。 - **noticeChan通道**:当有工作单元进入优先级队列排队后,会在通道里发送一个消息,以通知消费者worker从队列中获取元素(工作单元)进行消费。 - **消费者worker**:监听noticeChan,当监听到noticeChan有消息时,说明队列中有工作单元需要被处理,优先从高优先级队列中获取元素进行消费。 #### 02 队列-消费者模式 根据队列个数和消费者个数,我们可以将队列-消费者模式分为**单队列-单消费者模式**、**多队列(优先级队列)- 单消费者模式**、**多队列(优先级队列)- 多消费者模式**。 我们先从最简单的单队列-单消费者模式实现,然后一步步演化成多队列(优先级队列)-多消费者模式。 #### 03 单队列-单消费者模式实现 ![图2-单消费者模式.png](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/52a0b5e2e6da4345a6847a7f2c9d0475~tplv-k3u1fbpfcp-watermark.image?) ##### 3.1 队列的实现 我们先来看下队列的实现。这里我们用Golang中的List数据结果来实现,List数据结构是一个双向链表,包含了将元素放到链表尾部、将头部元素弹出的操作,符合队列**先进先出**的特性。 好,我们看下具体的队列的数据结构: ```golang type JobQueue struct { mu sync.Mutex //队列的操作需要并发安全 jobList *list.List //List是golang库的双向队列实现,每个元素都是一个job noticeChan chan struct{} //入队一个job就往该channel中放入一个消息,以供消费者消费 } ``` - **入队操作** ```golang /** * 队列的Push操作 */ func (queue *JobQueue) PushJob(job Job) { queue.jobList.PushBack(job) //将job加到队尾 queue.noticeChan <- struct{}{} } ``` 到这里有同学就会问了,**为什么不直接将job推送到Channel中,然后让消费者依次消费不就行了么?是的,单队列这样是可以的,因为我们最终目标是为了实现优先级的多队列,所以这里即使是单队列,我们也使用List数据结构,以便后续的演变**。 还有一点,大家注意到了,这里入队操作时有一个 这样的操作: ```golang queue.noticeChan <- struct{}{} ``` 消费者监听的实际上不是队列本身,而是通道noticeChan。当有一个元素入队时,就往noticeChan通道中输入一条消息,这里是一个空结构体,主要作用就是通知消费者worker,队列里有要处理的元素了,可以从队列中获取了。 这个在后面演化成多队列以及多消费者模式时会很有用。 - **出队操作** 根据队列的**先进先出**原则,是要获取队列的最先进入的元素。Golang中List结构体的Front()函数是获取链表的第一个元素,然后通过Remove函数将该元素从链表中移出,即得到了队列中的第一个元素。这里的Job结构体先不用关心,我们后面实现工作单元Job时,会详细讲解。 ```golang /** * 弹出队列的第一个元素 */ func (queue *JobQueue) PopJob() Job { queue.mu.Lock() defer queue.mu.Unlock() /** * 说明在队列中没有元素了 */ if queue.jobList.Len() == 0 { return nil } elements := queue.jobList.Front() //获取队里的第一个元素 return queue.jobList.Remove(elements).(Job) //将元素从队列中移除并返回 } ``` - **等待通知操作** 上面我们提到,消费者监听的是noticeChan通道。当有元素入队时,会往noticeChan中输入一条消息,以便通知消费者进行消费。如果队列中没有要消费的元素,那么消费者就会阻塞在该通道上。 ```golang func (queue *JobQueue) WaitJob() <-chan struct{} { return queue.noticeChan } ``` ##### 3.2 工作单元--Job的实现 一个工作单元就是一个要执行的任务。在系统中往往需要执行不同的任务,就是需要有不同类型的工作单元,但这些工作单元都有一组共同的执行流程。我们看下工作单元的类图。 ![图3-job类图.png](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/ab9840350b3c459e94dada1b79d5ec9e~tplv-k3u1fbpfcp-watermark.image?) 我们看下类图中的几个角色: - **Job接口**:定义了所有Job要实现的方法。 - **BaseJob类(结构体)**:定义了具体Job的基类。因为具体Job类中的有共同的属性和方法。所以抽象出一个基类,避免重复实现。但该基类对Execute方法没有实现,因为不同的工作单元有具体的执行逻辑。 - **SquareJob和AreaJob类(结构体)**:是我们要具体实现的业务工作Job。主要是实现Execute的具体执行逻辑。根据业务的需要定义自己的工作Job和对应的Execute方法即可。 接下来,我们以计算一个int类型数字的平方的SquareJob为例来看下具体的实现。 - **BaseJob结构体** 首先看下该结构体的定义 ```golang type BaseJob struct { Err error DoneChan chan struct{} //当作业完成时,或者作业被取消时,通知调用者 Ctx context.Context cancelFunc context.CancelFunc } ``` 在该结构体中,我们主要关注DoneChan字段就行,该字段是当具体的Job的Execute执行完成后,来通知调用者的。 再来看Done函数,该函数就是在Execute函数完成后,要关闭DoneChan通道,以解除Job的阻塞而继续执行其他逻辑。 ```golang /** * 作业执行完毕,关闭DoneChan,所有监听DoneChan的接收者都能收到关闭的信号 */ func (job *BaseJob) Done() { close(job.DoneChan) } ``` 再来看WaitDone函数,该函数是当Job执行后,要等待Job执行完成,在未完成之前,DoneChan里没有消息,通过该函数就能将job阻塞,直到Execute中调用了Done(),以便解除阻塞。 ```golang /** * 等待job执行完成 */ func (job *BaseJob) WaitDone() { select { case <-job.DoneChan: return } } ``` - **SquareJob结构体** ```golang type SquareJob struct { *BaseJob x int } ``` 从结构体的定义中可知,SquareJob嵌套了BaseJob,所以该结构体拥有BaseJob的所有字段和方法。在该结构体主要实现了Execute的逻辑:对x求平方。 ```golang func (s *SquareJob) Execute() error { result := s.x * s.x fmt.Println("the result is ", result) return nil } ``` ##### 3.3 消费者Worker的实现 Worker主要功能是通过监听队列里的noticeChan是否有需要处理的元素,如果有元素的话从队列里获取到要处理的元素job,然后执行job的Execute方法。 我们将该结构体定位为WorkerManager,因为在后面我们讲解多Worker模式时,会需要一个Worker的管理者,因此定义成了WorkerManager。 ```golang type WorkerManager struct { queue *JobQueue closeChan chan struct{} } ``` StartWorker函数,只有一个for循环,不断的从队列中获取Job。获取到Job后,进行消费Job,即ConsumeJob。 ```golang func (m *WorkerManager) StartWork() error { fmt.Println("Start to Work") for { select { case <-m.closeChan: return nil case <-m.queue.noticeChan: job := m.queue.PopJob() m.ConsumeJob(job) } } return nil } func (m *WorkerManager) ConsumeJob(job Job) { defer func() { job.Done() }() job.Execute() } ``` 到这里,单队列-单消费者模式中各角色的实现就讲解完了。我们通过main函数将其关联起来。 ```golang func main() { //初始化一个队列 queue := &JobQueue{ jobList: list.New(), noticeChan: make(chan struct{}, 10), } //初始化一个消费worker workerManger := NewWorkerManager(queue) // worker开始监听队列 go workerManger.StartWork() // 构造SquareJob job := &SquareJob{ BaseJob: &BaseJob{ DoneChan: make(chan struct{}, 1), }, x: 5, } //压入队列尾部 queue.PushJob(job) //等待job执行完成 job.WaitDone() print("The End") } ``` #### 04 多队列-单消费者模式 有了单队列-单消费者的基础,我们如何实现多队列-单消费者模式。也就是优先级队列。 ![图2-多队列-单消费者模式.png](https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/b57eb4c12994491588c8b4e0e6b9be72~tplv-k3u1fbpfcp-watermark.image?) 优先级的队列,实质上就是根据工作单元Job的优先级属性,将其放到对应的优先级队列中,以便worker可以根据优先级进行消费。我们要在Job结构体中增加一个Priority属性。因为该属性是所有Job都共有的,因此定义在BaseJob上更合适. ```golang type BaseJob struct { Err error DoneChan chan struct{} //当作业完成时,或者作业被取消时,通知调用者 Ctx context.Context cancelFunc context.CancelFunc priority int //工作单元的优先级 } ``` 我们再来看看多队列如何实现。实际上就是用一个切片来存储各个队列,切片的每个元素存储一个JobQueue队列元素即可。 ```golang var queues = make([]*JobQueue, 10, 100) ``` 那各优先级的队列在切片中是如何存储的呢?切片索引顺序只代表优先级的高于低,不代表具体是哪个优先级。 什么意思呢?假设我们现在对目前的工作单元定义了1、4、7三个优先级。这3个优先级在切片中是按优先级从小到到依次存储在queues切片中的,如下图: ![图4-正确的切片存储的优先级.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/305816ea901246a0a2eaf1b8745a3b01~tplv-k3u1fbpfcp-watermark.image?) 那为什么不让切片的索引就代表优先级,让优先级为1的队列存储在索引1处,优先级4的队列存储在索引4处,优先级7的队列存储在索引7处呢?如果这样存储的话,就会变成如下这样: ![图4-直接使用索引作为优先级缺点.png](https://p1-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/de69ec0c46a74d1b886a805f1630d99c~tplv-k3u1fbpfcp-watermark.image?) 可见如果我们设定的优先级不是连续的,那么就会造成空间的浪费。所以,我们是将队列按优先级高低依次存放到了切片中。 那既然这样,**当一个优先级的job来了之后,我该怎么知道该优先级的队列是存储在哪个索引中呢?我们用一个map来映射优先级和切片索引之间的关系。这样当一个工作单元Job入队的时候,以优先级为key,就可以查找到对应优先级的队列存储在切片的哪个位置了**。如下图所示: ![图5-优先级和索引映射.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/43fbb2ce760b4d1a938b199dd6fbdce8~tplv-k3u1fbpfcp-watermark.image?) 代码定义: ```golang var priorityIdx map[int][int]//该map的key是优先级,value代表的是queues切片的索引 ``` 好了,我们重新定义一下队列的结构体: ```golang type PriorityQueue struct { mu sync.Mutex noticeChan chan struct{} queues []*JobQueue priorityIdx map[int]int } //原来的JobQueue会变成如下这样: type JobQueue struct { priority int //代表该队列是哪种优先级的队列 jobList *list.List //List是golang库的双向队列实现,每个元素都是一个job } ``` 这里我们注意到有以下几个变化: - JobQueue里多了一个Priority属性,代表该队列是哪个优先级别。 - noticeChan属性从JobQueue中移动到了PriorityQueue中。因为现在有多个队列,只要任意一个队列里有元素就需要通知消费者worker进行消费,因此消费者worker监听的是PriorityQueue中是否有元素,而在监听阶段不关心具体哪个优先级队列中有元素。 好了,数据结构定义完了,我们看看将工作单元Job推入队列和从队列中弹出Job又有什么变化。 - **优先级队列的入队操作** 优先级队列的入队操作,就需要根据入队Job的优先级属性放到对应的优先级队列中,入队流程图如下: ![图6-优先级队列入队流程.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/9270c020d18e4cecb797cdc78761c365~tplv-k3u1fbpfcp-watermark.image?) 当一个Job加入队列的时候,有两种场景,一种是该优先级的队列已经存在,则直接Push到队尾即可。一种是该优先级的队列还不存在,则需要先创建该优先级的队列,然后再将该工作单元Push到队尾。如下是两种场景。 **队列已经存在的场景** 这种场景会比较简单。假设我们要插入优先级为7的工作单元,首先从映射表中查找7是否存在,发现对应关系是2,则直接找到切片中索引2的元素,即优先级为7的队列,将job加入即可。如下图。 ![图7-已存在队列插入.png](https://p1-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/a39bf251180547219db1fb7dfdc6b8f1~tplv-k3u1fbpfcp-watermark.image?) **队列不存在的场景** 这种场景稍微复杂些,在映射表中找不到要插入优先级的队列的话,则需要在切片中插入一个优先级队列,而为了优先级队列在切片中也保持有序(保持有序就可以知道队列的优先级的高低了),则需要移动相关的元素。我们以插入优先级为6的工作单元为例来讲解。 1、首先,我们的队列有一个初始化的状态,存储了优先级1、4、7的队列。如下图。 ![图7-优先级查找1.png](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/695e4bd875e947739a0396d938d0caca~tplv-k3u1fbpfcp-watermark.image?) 2、当插入优先级为6的工作单元时,发现在映射表中没有优先级6的映射关系,说明在切片中还没有优先级为6的队列的元素。所以需要在切片中依次查找到优先级6应该插入的位置在4和7之间,也就是需要存储在切片2的位置。 ![图7-优先级查找2.png](https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/13f72988d9a24c2498705b42b8686b64~tplv-k3u1fbpfcp-watermark.image?) 3、将原来索引2位置的优先级为7的队列往后移动到3,同时更新映射表中的对应关系。 ![图7-优先级查找3.png](https://p1-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/4bb2184a8d764d3f9699c185e371b74a~tplv-k3u1fbpfcp-watermark.image?) 4、将优先级为6的工作单元插入到索引2的队列中,同时更新映射表中的优先级和索引的关系。 ![图7-优先级查找4.png](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/320de088327e4efbbda436861c2dbcbf~tplv-k3u1fbpfcp-watermark.image?) 我们看下代码实现: ```golang func (priorityQueue *PriorityQueue) Push(job Job) { priorityQueue.mu.Lock() defer priorityQueue.mu.Unlock() //先根据job的优先级找要入队的队列 var idx int var ok bool //从优先级-切片索引的map中查找该优先级的队列是否存在 if idx, ok = priorityQueue.priorityIdx[job.Priority()]; !ok { //如果不存在该优先级的队列,则需要初始化一个队列,并返回该队列在切片中的索引位置 idx = priorityQueue.addPriorityQueue(job.Priority) } //根据获取到的切片索引idx,找到具体的队列 queue := priority.queues[idx] //将job推送到队列的队尾 queue.JobList.PushBack(job) //队列job个数+1 priorityQueue.Size++ //如果队列job个数超过队列的最大容量,则从优先级最低的队列中移除工作单元 if priorityQueue.size > priorityQueue.capacity { priorityQueue.RemoveLeastPriorityJob() }else { //通知新进来一个job priorityQueue.noticeChan <- struct{}{} } } ``` 代码中大部分也都做了注释,不难理解。这里我们来看下addPriorityQueue的具体实现: ```golang func (priorityQueue *PriorityQueue) addPriorityQueue(priority int) int { n := len(priorityQueue.queues) //通过二分查找找到priority应插入的切片索引 pos := sort.Search(n, func(i int) bool { return priority < priorityQueue.priority }) //更新映射表中优先级和切片索引的对应关系 for i := pos; i < n; i++ { priorityQueue.priorityIdx[priorityQueue.queues[i].priority] = i + 1 } tail := make([]*jobQueue, n-pos) copy(tail, priorityQueue.queues[pos:]) //初始化一个新的优先级队列,并将该元素放到切片的pos位置中 priorityQueue.queues = append(priorityQueue.queues[0:pos], newJobQueue(priority)) //将高于priority优先级的元素也拼接到切片后面 priorityQueue.queues = append(priorityQueue.queues, tail...) return pos } ``` 最后,我们再来看一个实际的调用例子: ```golang func main() { //初始化一个队列 queue := &PriorityQueue{ noticeChan: make(chan struct{}, cap), capacity: cap, priorityIdx: make(map[int]int), size: 0, } //初始化一个消费worker workerManger := NewWorkerManager(queue) // worker开始监听队列 go workerManger.StartWork() // 构造SquareJob job := &SquareJob{ BaseJob: &BaseJob{ DoneChan: make(chan struct{}, 1), }, x: 5, priority: 10, } //压入队列尾部 queue.PushJob(job) //等待job执行完成 job.WaitDone() print("The End") } ``` #### 05 多队列-多消费者模式 ![图2-多队列-多消费者模式.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/cb2d7fd3da1b4de488bc0c39add9539e~tplv-k3u1fbpfcp-watermark.image?) 我们在多队列-单消费者的基础上,再来看看多消费者模式。也就是增加worker的数量,提高Job的处理速度。 我们再来看下worker的定义: ```golang type WorkerManager struct { queue *PriorityQueue closeChans []chan struct{} } ``` 这里需要注意,closeChans变成了切片数组。因为我们每启动一个worker,就需要有一个关闭通道。 然后看StartWorker函数的实现: ```golang func (m *WorkerManager) StartWork(n int) error { fmt.Println("Start to Work") for i := 0; i < n; i++ { m.createWorker(); } return nil } func (m *WorkerManager) createWorker() { closeChan := make(chan struct{}) //每个协程,就是一个worker go func(closeChan chan struct{}) { var job Job for { select { case <-m.closeChan: return nil case <-m.queue.noticeChan: job := m.queue.PopJob() m.ConsumeJob(job) } } }(closeChan) m.closeChanMu.Lock() defer m.closeChanMu.Unlock() m.closeChans = append(m.closeChans, closeChan) return nil } func (m *WorkerManager) ConsumeJob(job Job) { defer func() { job.Done() }() job.Execute() } ``` 这里需要注意的是,所有的worker都需要监听队列的noticeChan通道。测试的例子就留给读者自己了。 另外如下图的单队列-多消费者模式是多队列-多消费者模式的一个特例,这里就不再进行实现了。 ![图2-单队列-多消费者模式.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/dbee73190ef1490d84a35fa2d951d9d8~tplv-k3u1fbpfcp-watermark.image?) ### 总结 队列的作用可以用来控制流量,而优先级队列在兼顾流量控制的同时,还能将流量按优先级高低来进行处理。 本文中一些细节的并发加锁操作做了忽略,大家在实际应用中根据需要进行完善即可。

有疑问加站长微信联系(非本文作者))

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

679 次点击  ∙  2 赞  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传