likes
comments
collection
share

从理论到实践：Queue 和 Simple Queue 的详细解析

作者站长头像
站长
· 阅读数 2

我是 LEE，老李，一个在 IT 行业摸爬滚打 17 年的技术老兵。

这篇是 WorkQueue 项目介绍的第二篇文章，上一篇是《简约而不简单：WorkQueue 的轻量级高效之道》。在上一篇文章中，我们介绍了 WorkQueue 的基本原理和架构，但是我还没有介绍 WorkQueue 核心模块 Queue 和 Simple Queue 的实现原理。本篇文章就是要介绍这两个模块的实现原理，以及如何使用它们。

事件背景

在日常开发工作中，经常会使用 Queue 的数据结构，比如在消息队列中，我们会使用 Queue 来存储消息，然后再从 Queue 中取出消息进行处理。在 Kubernetes 中，也有很多地方使用了 Queue，比如 Controller 中的 WorkQueue，Scheduler 中的 SchedulingQueue，Kubelet 中的 PodWorkers 等等。在 Kubernetes 中，Queue 的使用场景非常多，都是基于 FIFO 的队列，只是在具体的实现上有所差异。Queue 在我们日常开发使用中也是一个非常常见的组件，甚至可以说是必不可少的组件。

如果从数据存储和使用角度中，可以将 Queue 中的数据分为两种方式：

  • 可以重复：Queue 中的数据可以重复，比如应用在不同时间产生的两条内容相同的日志。 WorkQueue 中使用了 Simple Queue 作为实现。
  • 保证唯一：Queue 中的数据保证唯一，比如消息队列中的消息，每条消息和事件都是唯一的。 WorkQueue 中使用了 Queue 作为实现

Queue 介绍

Queue 和 Simple Queue 的逻辑结构是一样的，都是基于 FIFO 的队列，只是在具体的实现上有所差异。Queue 的实现是基于 set 和 deque 的，Simple Queue 的实现是基于 deque 的。通俗的说：Queue 不运行存储的数据有重复，而 Simple Queue 可以存储重复的数据。

1. Queue

1.1 Queue 简介

Queue 是一个 FIFO 的队列，基于标准的 Interface 实现所有的功能接口，包括 Add，Get，Len，GetWithBlock，Done，Stop，IsClosed。Queue 使用 set 来保证数据的唯一性，deque 用来按顺序存储数据。Queue 实现非常简单。

// 队列方法接口
// Queue interface
type Interface interface {
	Add(element any) error // 添加元素
	Len() int // 队列长度
	Get() (element any, err error) // 获取元素
	GetWithBlock() (element any, err error)  // 阻塞获取元素
	Done(element any)  // 完成处理
	Stop() // 停止队列
	IsClosed() bool // 队列是否关闭
}

Queue 在保证数据唯一上采用了 processing 和 dirty 是两个 set，processing 用来存储正在处理的数据，dirty 用来存储等待理完成的数据。processing 和 dirty 两个 set 的数据都是唯一的，Queue 的 deque 中存储的数据是 dirty + processing 的数据。

Queue 在数据处理的生命周期中还包含可以被自己定义的 Callback 方法，方便使用者干预或者介入数据处理的生命周期。Queue 的 Callback 方法包括 OnAdd，OnGet，OnDone。

// 队列的回调接口
// Callback interface
type Callback interface {
	OnAdd(any) // 添加元素回调
	OnGet(any) // 获取元素回调
	OnDone(any) // 完成处理回调
}

1.2 Queue 实现

Queue 整体结构设计如下：

从理论到实践：Queue 和 Simple Queue 的详细解析

1.3 Queue 使用举例

当然 Queue 使用起来也非常简单，没有太多复杂的参数初始化过程。 但是这里有一个非常需要注意的地方： 不论使用 Get 还是 GetWithBlock 之后一定要记得使用 Done 方法，来标记这个数据已经处理完成，否则再往 Queue 添加相同的数据的时候，会返回 ErrorQueueElementExist 错误。

代码举例

package main

import (
	"fmt"
	"time"

	"github.com/shengyanli1982/workqueue"
)

func main() {
	q := workqueue.NewQueue(nil) // create a queue

	go func() {
		for {
			element, err := q.Get() // get element from queue
			if err != nil {
				fmt.Println(err)
				return
			}
			fmt.Println("get element:", element)
			q.Done(element) // mark element as done, 'Done' is required after 'Get'
		}
	}()

	_ = q.Add("hello") // add element to queue
	_ = q.Add("world")

	time.Sleep(time.Second * 2) // wait for element to be executed

	q.Stop()
}

如果你不想使用 workqueue.NewQueue(nil) 这个方式来创建队列，还有一个函数偷懒 workqueue.DefaultQueue()，两者是等效的。

如果你想在 Queue 过程中使用 Callback 方法，这个使用的方法：

// 实现 Callback 接口
type callback struct {}

func (cb *callback) OnAdd(item any) {}
func (cb *callback) OnGet(item any) {}
func (cb *callback) OnDone(item any) {}

// 创建 Config
conf := NewQConfig()
// 关联 Callback
conf.WithCallback(&callback{})

// 创建 Queue 的时候传入这个 conf 对象，而不是在用 nil
q := NewQueue(conf)

后续代码跟上面的 代码举例 中的内容一样。

2.4 Queue 代码解析

Add

// 添加元素到队列
// Add element to queue
func (q *Q) Add(element any) error {
	if q.IsClosed() {
		// 队列已经关闭，返回 ErrorQueueClosed 错误
		return ErrorQueueClosed
	}

	if q.isElementMarked(element) {
		// 判断元素是否已经被标记，如果已经被标记，返回 ErrorQueueElementExist 错误
		// 判断元素是否在 processing 和 dirty 中
		return ErrorQueueElementExist
	}

	// 从 nodepool 中获取一个 node，node 是一个双向链表的节点，nodepool 是一个 sync.Pool
	n := q.nodepool.Get()
	n.SetData(element)

	q.cond.L.Lock()
	q.queue.Push(n) // 将 node 添加到 queue 中
	q.cond.Signal() // 发送信号，通过 sync.Cond 通知等待的方法
	q.cond.L.Unlock()

	q.prepare(element) // 将数据放入 dirty 中
	q.config.cb.OnAdd(element) // 执行添加回调

	return nil
}

Get

// 从队列中获取一个元素, 如果队列为空，不阻塞等待
// Get an element from the queue.
func (q *Q) Get() (element any, err error) {
	if q.IsClosed() {
		// 队列已经关闭，返回 ErrorQueueClosed 错误
		return nil, ErrorQueueClosed
	}

	q.qlock.Lock()
	n := q.queue.Pop() // 从 queue 中获取一个 node
	q.qlock.Unlock()
	if n == nil {
		// 如果 node 为空，返回 ErrorQueueEmpty 错误, 这里不会阻塞等待
		return nil, ErrorQueueEmpty
	}

	element = n.Data() // 获取 node 中的数据
	q.todo(element) // 将数据从 dirty 删除，放入 processing 中
	q.config.cb.OnGet(element) // 执行获取回调
	q.nodepool.Put(n) // 将 node 放回 nodepool 中

	return element, nil
}

GetWithBlock

// 从队列中获取一个元素，如果队列为空，阻塞等待
// Get an element from the queue, if the queue is empty, block and wait.
func (q *Q) GetWithBlock() (element any, err error) {
	if q.IsClosed() {
		return nil, ErrorQueueClosed
	}

	q.cond.L.Lock()
	for q.queue.Len() == 0 { // 如果 queue 为空，阻塞等待
		q.cond.Wait() // 等待信号
	}
	n := q.queue.Pop()
	q.cond.L.Unlock()
	if n == nil {
		return nil, ErrorQueueEmpty
	}

	element = n.Data()
	q.todo(element)
	q.config.cb.OnGet(element)
	q.nodepool.Put(n)

	return element, nil
}

2. Simple Queue

2.1 Simple Queue 简介

Simple Queue 跟 Queue 十分类似，代码也大量复用。Simple Queue 是一个 FIFO 的队列，基于标准的 Interface 实现所有的功能接口，包括 Add，Get，Len，GetWithBlock，Done，Stop，IsClosed。Simple Queue 使用 deque 来存储数据，deque，但是没有 set 数据结构，也是说 Simple Queue 中的数据可以重复。

// 队列方法接口
// Queue interface
type Interface interface {
	Add(element any) error // 添加元素
	Len() int // 队列长度
	Get() (element any, err error) // 获取元素
	GetWithBlock() (element any, err error)  // 阻塞获取元素
	Done(element any)  // 完成处理
	Stop() // 停止队列
	IsClosed() bool // 队列是否关闭
}

Simple Queue 在数据处理的生命周期中还包含可以被自己定义的 Callback 方法，方便使用者干预或者介入数据处理的生命周期。Simple Queue 的 Callback 方法包括 OnAdd，OnGet，OnDone。

// 队列的回调接口
// Callback interface
type Callback interface {
	OnAdd(any) // 添加元素回调
	OnGet(any) // 获取元素回调
	OnDone(any) // 完成处理回调
}

2.2 Simple Queue 实现

Simple Queue 整体结构设计如下：

从理论到实践：Queue 和 Simple Queue 的详细解析

2.3 Simple Queue 使用举例

当然 Queue 使用起来也非常简单，没有太多复杂的初始化参数过程。 但是这里有一个非常需要注意的地方： 这里 Get 还是 GetWithBlock 可以不使用 Done 方法。Done 方法，是一个空壳方法，没有任何实际的逻辑，只是为了保持 Simple Queue 和 Queue 的接口一致。Simple Queue 中的数据可以重复，所以不需要使用 Done 方法来标记数据已经处理完成。

代码举例

package main

import (
	"fmt"
	"time"

	"github.com/shengyanli1982/workqueue"
)

func main() {
	q := workqueue.NewSimpleQueue(nil) // create a queue

	go func() {
		for {
			element, err := q.Get() // get element from queue
			if err != nil {
				fmt.Println(err)
				return
			}
			fmt.Println("get element:", element)
			q.Done(element) // mark element as done, 'Done' is required after 'Get'
		}
	}()

	_ = q.Add("hello") // add element to queue
	_ = q.Add("world")

	time.Sleep(time.Second * 2) // wait for element to be executed

	q.Stop()
}

如果你不想使用 workqueue.NewSimpleQueue(nil) 这个方式来创建队列，还有一个函数偷懒 workqueue.DefaultSimpleQueue()，两者是等效的。

如果你想在 Simple Queue 过程中使用 Callback 方法，这个使用的方法：

// 实现 Callback 接口
type callback struct {}

func (cb *callback) OnAdd(item any) {}
func (cb *callback) OnGet(item any) {}
func (cb *callback) OnDone(item any) {}

// 创建 Config
conf := NewQConfig()
// 关联 Callback
conf.WithCallback(&callback{})

// 创建 Simple Queue 的时候传入这个 conf 对象，而不是在用 nil
q := NewSimpleQueue(conf)

后续代码跟上面的 代码举例 中的内容一样。

2.4 Simple Queue 代码解析

Add

// 添加元素到队列
// Add element to queue
func (q *SimpleQ) Add(element any) error {
	if q.IsClosed() {
		// 队列已经关闭，返回 ErrorQueueClosed 错误
		return ErrorQueueClosed
	}

	// 对比 Queue 没有操作 set 相关的部分

	// 从 nodepool 中获取一个 node，node 是一个双向链表的节点，nodepool 是一个 sync.Pool
	n := q.nodepool.Get()
	n.SetData(element)

	q.cond.L.Lock()
	q.queue.Push(n) // 将 node 添加到 queue 中
	q.cond.Signal() // 发送信号，通过 sync.Cond 通知等待的方法
	q.cond.L.Unlock()

	// 对比 Queue 没有操作 set 相关的部分

	q.config.cb.OnAdd(element) // 执行添加回调

	return nil
}

Get

// 从队列中获取一个元素, 如果队列为空，不阻塞等待
// Get an element from the queue.
func (q *SimpleQ) Get() (element any, err error) {
	if q.IsClosed() {
		return nil, ErrorQueueClosed
	}

	q.qlock.Lock()
	n := q.queue.Pop()
	q.qlock.Unlock()
	if n == nil { // 队列为空 (queue is empty)
		return nil, ErrorQueueEmpty
	}

	element = n.Data() // 获取 node 中的数据
	q.config.cb.OnGet(element) // 执行获取回调
	q.nodepool.Put(n) // 将 node 放回 nodepool 中

	// 对比 Queue 没有操作 set 相关的部分

	return element, nil
}

GetWithBlock

// 从队列中获取一个元素，如果队列为空，阻塞等待
// Get an element from the queue, if the queue is empty, block and wait.
func (q *Q) GetWithBlock() (element any, err error) {
	if q.IsClosed() {
		return nil, ErrorQueueClosed
	}

	q.cond.L.Lock()
	for q.queue.Len() == 0 { // 如果 queue 为空，阻塞等待
		q.cond.Wait() // 等待信号
	}
	n := q.queue.Pop()
	q.cond.L.Unlock()
	if n == nil {
		return nil, ErrorQueueEmpty
	}

	element = n.Data()
	q.config.cb.OnGet(element)
	q.nodepool.Put(n)

	// 对比 Queue 没有操作 set 相关的部分

	return element, nil
}

总结

谢谢你，到此你已经看完了两篇文章了。通过本篇文章向你介绍了 Queue 还是 Simple Queue 的定义、使用方法和实现原理。 希望能够让你对 Queue 还是 Simple Queue 有一个清晰的认识。

Queue 和 Simple Queue 作为 WorkQueue 的核心模块，它们的实现非常简单，也非常高效，可以满足大部分的使用场景。 在实际的使用过程中，还是需要根据实际的业务场景来选择使用 Queue 还是 Simple Queue。如果你的数据需要保证唯一性，那么你就需要使用 Queue，如果你的数据可以重复，那么你就可以使用 Simple Queue。

下一篇将讲围绕介绍 Delaying Queue 和 Priority Queue 模块，包括使用方法和代码讲解。这两个高级的 Queue 都是基于 Queue 和 Simple Queue 实现的，所以你对 Queue 和 Simple Queue 有了清晰的认识之后，对 Delaying Queue 和 Priority Queue 的理解就会非常容易了。