Go 語言實戰：構建強大的延遲任務隊列
01 介紹
延遲隊列是一種數據結構，用於處理需要在未來某個特定時間執行的任務。這些任務被添加到隊列中，並且指定了一個執行時間，只有到達指定的時間點時才能從隊列中取出並執行。
在實際應用中，延遲隊列可以用於處理各種需要延遲處理的任務，例如發送郵件提醒、訂單自動取消、對超時任務的處理等。由於任務的執行是在未來的某個時間點，因此這些任務不會立即執行，而是存儲在隊列中，直到它的預定執行時間纔會被執行。
02 Simple
在 Go 語言中，我們可以使用 time 包提供的計時器功能，通過使用 Go 中的 slice 存儲延遲處理的任務，實現一個簡單的延遲隊列的功能。
示例代碼：
type Task struct {
ExecuteTime time.Time
Job func()
}
首先，我們定義一個結構體 Task，它包含一個可以執行任務的函數 Job，和一個執行時間 ExecuteTime，這是期望執行該函數的時間。
示例代碼：
type DelayQueue struct {
TaskQueue []Task
}
接下來，我們定義一個 DelayQueue 結構體，它擁有一個 TaskQueue，這是一個 Task 類型的切片，用於保存待執行任務的列表。
示例代碼：
// 添加任務
func (d *DelayQueue) AddTask(t Task) {
d.TaskQueue = append(d.TaskQueue, t)
}
// 移除任務
func (d *DelayQueue) RemoveTask() {
d.TaskQueue = d.TaskQueue[1:]
}
// 執行任務
func (d *DelayQueue) ExecuteTasks() {
for len(d.TaskQueue) > 0 {
// 獲取隊列最頂部的任務
currentTask := d.TaskQueue[0]
// 如果執行時間還沒到，等待
if time.Now().Before(currentTask.ExecuteTime) {
time.Sleep(currentTask.ExecuteTime.Sub(time.Now()))
}
// 執行任務
currentTask.Job()
// 移除已執行的任務
d.RemoveTask()
}
}
DelayQueue 包含三個方法：
第一個方法是 AddTask(t Task)。此方法將提供的任務 t 添加到 TaskQueue 的末尾。
第二個方法是 RemoveTask()。此方法從 TaskQueue 中移除第一個任務。
第三個方法是 ExecuteTasks()。此方法將執行 TaskQueue 中的所有任務。如果隊列頂部任務的執行時間還未到，該方法將等待。一旦時間到了，它將會執行 Job 並從 TaskQueue 中移除該任務。
示例代碼：
func main() {
fmt.Println("Start DelayQueue")
queue := DelayQueue{}
firstTask := Task{
ExecuteTime: time.Now().Add(4 * time.Second),
Job: func() {
fmt.Println("Executed task 1 after delay")
},
}
queue.AddTask(firstTask)
secondTask := Task{
ExecuteTime: time.Now().Add(10 * time.Second),
Job: func() {
fmt.Println("Executed task 2 after delay")
},
}
queue.AddTask(secondTask)
queue.ExecuteTasks()
fmt.Println("Done!")
}
輸出結果：
Start DelayQueue
Executed task 1 after delay
Executed task 2 after delay
Done!在示例代碼中，我們創建了一個延時隊列，將任務添加到隊列中，並在指定的延時後執行它們。
通過使用這些結構體和方法，我們可以在 Go 中實現簡單的延遲執行任務的功能。
但是，當 Go 程序重啓時，存儲在 slice 中的延遲處理的任務將全部丟失。
03 Complex
在 Go 程序中，如果想在重啓後保留數據，我們可以將數據持久化到 Redis，可以使用 go-redis/redis 庫 [1] 與 Redis 交互。而對於延遲隊列的需求，則可以使用 Redis 的 ZSET（有序集合）特性來實現。
示例代碼：
// 定義一個全局的redisdb變量
var redisdb *redis.Client
// 初始化連接
func initClient() (err error) {
redisdb = redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "", // no password set
DB: 0, // use default DB
})
_, err = redisdb.Ping().Result()
if err != nil {
return err
}
return nil
}
全局變量 redisdb 是 redis.Client 類型的指針，用來保存到 Redis 客戶端的引用。
initClient 函數初始化連接到 Redis 服務器，該服務器在本地主機的 6379 端口運行。它將一個新的 Redis 客戶端分配給 redisdb 變量。如果連接成功，它就會 ping Redis 服務器以測試連接。
示例代碼：
// 向隊列中添加任務
func addTaskToQueue(task string, executeTime int64) {
err := redisdb.ZAdd("delay-queue", redis.Z{
Score: float64(executeTime),
Member: task,
}).Err()
if err != nil {
panic(err)
}
}
addTaskToQueue 函數將具有執行時間的任務添加到 Redis 等待排序的集合 "delay-queue"。執行時間是一個 UNIX 時間戳，作爲排序集合中的項目的 score，允許 Redis 按照他們應該執行的時間來排序項目。
示例代碼：
// 從隊列中獲取並處理任務
func getAndExecuteTasks() {
for {
// 使用 ZRANGEBYSCORE 命令獲取分數（時間戳）<= 當前時間的任務
tasks, err := redisdb.ZRangeByScore("delay-queue", redis.ZRangeBy{
Min: "-inf",
Max: fmt.Sprintf("%d", time.Now().Unix()),
}).Result()
if err != nil {
time.Sleep(1 * time.Second)
continue
}
// 處理任務
for _, task := range tasks {
fmt.Println("Executing task: ", task)
// 執行完任務後，用 ZREM 移除該任務
redisdb.ZRem("delay-queue", task)
}
// 暫停一秒
time.Sleep(1 * time.Second)
}
}
getAndExecuteTasks 函數不斷檢查 "delay-queue"。它提取隊列中 score 小於或等於當前時間戳的任務，意味着這些任務現在應該執行或者他們應該在過去就已經執行。獲取任務後，它打印任務（模擬執行）並從隊列中刪除任務。
示例代碼：
func main() {
err := initClient()
if err != nil {
fmt.Println("redis connect error:", err)
return
}
// 添加一些測試任務
addTaskToQueue("task1", time.Now().Add(10*time.Second).Unix())
addTaskToQueue("task2", time.Now().Add(20*time.Second).Unix())
// 執行延遲隊列中的任務
getAndExecuteTasks()
}
輸出結果：
Executing task: task1
Executing task: task2main 函數調用這些函數。首先，它初始化 Redis 客戶端。如果初始化和連接成功，它將一些測試任務添加到隊列中，並啓動任務執行循環。
總結一下，這段 Go 代碼使用 Redis 的 Sorted Set 數據類型創建了一個延時隊列系統，其中的任務按照他們的執行時間進行排序，一個任務工作者循環獲取並執行隊列中的任務。這是一個簡單而高效地實現作業調度系統的方法。
04 總結
本文我們分別實現簡單版和複雜版的延遲隊列，其中簡單版延遲隊列，只使用 Go 實現，複雜版延遲隊列，使用 Go 和 Redis 實現。
只使用 Go 實現延遲隊列：
優點：
不需要外部依賴：只使用 Go 實現延遲隊列，你不需要安裝和維護外部的 Redis 服務器。
缺點：
健壯性和持久性：如果程序崩潰或重新啓動，延遲隊列的數據可能會丟失。
併發控制：使用 Go 內置的數據結構（如 channels 或 slices）在多個 goroutines 之間共享狀態變量可能需要精細的併發控制，比如使用 mutexes 或者 channels。
使用 Go + Redis 實現延遲隊列：
優點：
數據持久性：Redis 提供了數據持久性，即使在程序重啓或崩潰後，隊列中的數據依然可以恢復。
簡化併發：Redis 提供的數據結構（如 sorted set）是原子操作，可以簡化併發控制。
功能強大：使用 Redis，你可以利用其提供的一些內建功能，如超時、TTL、持久化等。
缺點：
需要額外的組件：使用 Redis 意味着需要安裝和運行 Redis 服務器，這可能增加系統的複雜性和運維成本。
網絡延遲：如果 Go 程序和 Redis 服務器不在同一臺機器上，網絡延遲可能會影響延遲的準確性。
總的來說，如果我們對延遲隊列的持久性、準確性和併發性有高要求，那麼 Go + Redis 的方案可能會更適合。如果我們想要一個更簡單的解決方案，並且可以容忍在程序崩潰時部分數據丟失，那麼只使用 Go 實現可能會更合適。
參考資料
[1] go-redis/redis 庫: https://redis.uptrace.dev/
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/OkBqiMEnha3SyCxBA0MkRQ