PHP轉Go系列 | ThinkPHP與Gin框架之Redis延時消息隊列技術實踐

我們在某寶或某多多上搶購商品時，如果只是下了訂單但沒有進行實際的支付，那在訂單頁面會有一個支付倒計時，要是過了這個時間點那麼訂單便會自動取消。在這樣的業務場景中，一般情況下就會使用到延時隊列。 ...

大家好，我是碼農先森。

我們在某寶或某多多上搶購商品時，如果只是下了訂單但沒有進行實際的支付，那在訂單頁面會有一個支付倒計時，要是過了這個時間點那麼訂單便會自動取消。在這樣的業務場景中，一般情況下就會使用到延時隊列。

通常在客戶下單之後，就會將訂單數據推送到延時隊列中並且會對該消息設置一個延時時長，比如設置五分鐘、十分鐘、或十五分鐘等，具體的時長應該還是要結合當前的業務進行衡量，然後消費端會在指定時間到達後就對該消息進行支付支付狀態判斷，如果已經支付則不予處理，要還是未支付，則會取消該訂單，並且釋放商品庫存。

我們這次分享的內容，主要是基於 Redis 延時隊列的實現方式，當然除了 Redis 還可以用其他的技術，比如 RabbitMQ、Kafka、RocketMQ 等專業的消息隊列。但是我用 Redis 的原因是，它的應用場景比較廣泛，我們平時接觸也比較多，而且相對於專業的消息隊列它沒有過多複雜的配置，學起來容易上手，出了問題解決起來也快，學東西的路徑都是由易到難嘛。

另外，如果你對上面提到的專業消息隊列使用很熟練，也可以將 Redis 更換成它們，這裡只是存儲介質的不同，技術的實現邏輯上沒有太大區別，重要的是設計思想，大家各取所需吧。

好了，我先介紹一下這次延時隊列的實現邏輯。主要分為三個部分，一是：消息的發送，如果設置了延時時間則會將消息存儲到 Redis 的延時隊列中，反之會直接將消息推送到 Redis 的就緒隊列中等待消費。二是：將到期的消息從 Redis 延時隊列中取出，並且推送到 Redis 的就緒隊列中等待消費。三是：消費端會從 Redis 的就緒隊列中按順序讀取出消息，並且執行對應的業務處理邏輯，如果處理失敗則會將該消息，再次推送到 Redis 的延時隊列中進行下一次的重試。

這裡說到的延時隊列是利用 Redis 有序集合來實現的，它每間隔一秒鐘就會被輪詢一次，如果有到期的消息，則就會將該消息推送到 Redis 就緒隊列，並且從該集合中移除過期的消息，至此就可以等待著消費端進行消費了。接下來我們就從實際的代碼出發，來看一下如何實現基於 Redis 的延時隊列。

話不多說，開整！我們先來看一下整體的項目目錄結構，內容主要分為 PHP 和 Go 兩部分。

[manongsen@root php_to_go]$ tree -L 2
.
├── go_delay
│   ├── app
│   │   ├── controller
│   │   │   └── notify.go
│   │   ├── config
│   │   │   └── config.go
│   │   ├── extend
│   │   │   └── queue.go
│   │   └── route.go
│   ├── go.mod
│   ├── go.sum
│   └── main.go
└── php_delay
│   ├── app
│   │   ├── controller
│   │   │   └── Notify.php
│   ├── composer.json
│   ├── composer.lock
│   ├── command
│   │   └── Consumer.php
│   ├── route
│   │   └── app.php
│   ├── extend
│   │   └── Queue.php
│   ├── think
│   ├── vendor
│   └── .env

ThinkPHP

使用 composer 創建基於 ThinkPHP 框架的 php_delay 項目。

## 當前目錄
[manongsen@root ~]$ pwd
/home/manongsen/workspace/php_to_go/php_delay

## 安裝 ThinkPHP 框架
[manongsen@root php_delay]$ composer create-project topthink/think php_delay
[manongsen@root php_delay]$ cp .example.env .env

## 安裝 Composer 依賴包
[manongsen@root php_delay]$ composer require predis/predis
## 創建一個消費者腳本
[manongsen@root php_delay]$ php think make:command Consumer
## 創建一個生產者腳本，用於測試
[manongsen@root php_delay]$ php think make:command Producer

這個就是延時隊列實現的核心類，定義了就緒、延時、失敗三個消息隊列。send() 方法用於發送消息，其中可以指定 $delay 參數設置延時時間單位是秒。wait() 方法用於消費端監聽消息，從下麵的代碼可以看出這裡還利用多進程，父進程的作用是每間隔一秒鐘，就從 Redis 有序集合中讀取到期的消息，並將該消息推送到 Redis 就緒隊列，子進程則阻塞監聽就緒隊列的消息，並且將接收到的消息回調到用戶自定義的業務函數中。

<?php
declare (strict_types = 1);

class Queue
{
    // 就緒消息存放的隊列
    const QUEUE_READY = 'redis:queue:ready'; 

    // 延遲消息存放的隊列(實際的數據結構是有序集合)
    const QUEUE_DELAY = 'redis:queue:delay'; 

    // 失敗消息存放的隊列
    const QUEUE_FAILED = 'redis:queue:failed'; 

    protected $_client;
    protected $_options = [
        'retry_seconds' => 5, // 重試延時5秒
        'max_attempts'  => 5, // 最大重試次數
    ];

    public function __construct()
    {
        // 與 Redis 建立連接
        $this->_client = new \think\cache\driver\Redis(config('cache.stores.redis'));
        $this->_client->get("ping");
    }

    // 發送消息
    public function send($data, $delay = 0)
    {
        static $_id = 0;
        $id = \microtime(true) . '.' . (++$_id);
        $now = time();
        $package_str = \json_encode([
            'id'       => $id,    // 消息ID
            'time'     => $now,   // 當前時間
            'delay'    => $delay, // 延遲時長(秒)
            'attempts' => 0,      // 重試次數
            'data'     => $data   // 消息內容
        ]);

        // 如果不是延時消息，則直接將消息推送到就緒隊列
        if ($delay == 0) {
            $this->_client->lpush(static::QUEUE_READY, $package_str);
        } else {
            // 否則將消息寫入到有序集合中
            $this->_client->zadd(static::QUEUE_DELAY, $now + $delay, $package_str);
        }
    }

    // 從有序集合中取出數據推送到就緒隊列中
    public function tryToPullDelayQueue()
    {
        while (true) {
            try {
                $now = time(); // 當前時間
                $options = ['LIMIT', 0, 128]; // 每次取 128 條數據
                $items = $this->_client->zrevrangebyscore(static::QUEUE_DELAY, $now, '-inf', $options);
                foreach ($items as $package_str) {
                    // 從有序集合中移除該數據
                    $result = $this->_client->zrem(static::QUEUE_DELAY, $package_str);
                    if ($result !== 1) {
                        continue;
                    }
                    // 將數據JSON反序列化解析
                    $package = \json_decode($package_str, true);
                    if (!$package) {
                        // 解析失敗則推送到失敗隊列
                        $this->_client->lpush(static::QUEUE_FAILED, $package_str);
                        continue;
                    }
                    // 將數據推送到就緒隊列
                    $this->_client->lpush(static::QUEUE_READY, $package_str);
                }
            } catch (\Throwable $e) {
                echo $e->getMessage() . PHP_EOL;
            }

            // 間隔1s之後再次輪詢
            sleep(1);
        }
    }

    // 監聽消息
    public function wait($success_callback, $failure_callback)
    {
        echo "開始監聽消息..." . PHP_EOL;
        // 創建一個進程
        // 父進程用於輪詢有序集合消息
        // 子進程監聽就緒隊列消息
        $pid = pcntl_fork();
        if ($pid < 0) {
            exit('fork error');
        } else if($pid > 0) {
            // 輪詢有序集合消息並推送到就緒隊列
            (new \Queue())->tryToPullDelayQueue();
            pcntl_wait($status);
            exit();
        }

        while (true) {
            try {            
                // 阻塞監聽就緒隊列消息
                $data = $this->_client->brpop(static::QUEUE_READY, 0);
                if ($data) {
                    $package_str = $data[1];
                    // 將數據JSON反序列化解析
                    $package = json_decode($package_str, true);
                    if (!$package) {
                        // 解析失敗則推送到失敗隊列
                        $this->_client->lpush(static::QUEUE_FAILED, $package_str);
                    } else {
                        try {
                            // 將消息回調到我們在業務層面寫的回調函數中
                            \call_user_func($success_callback, $package['data']);
                        } catch (\Throwable $e) {
                            $package['max_attempts'] = $this->_options['max_attempts'];
                            $package['error'] = $e->getMessage();
                            $package_modified = null;
                            // 如果出現異常並且我們設置了失敗回調函數
                            if ($failure_callback) {
                                try {
                                    // 則會回調到我們在業務層面寫的回調函數中
                                    $package_modified = \call_user_func($failure_callback, $e, $package);
                                } catch (\Throwable $ta) {
                                }
                            }
                            // 如果修改了消息內容，則重新構造消息
                            if (is_array($package_modified)) {
                                $package['data'] = $package_modified['data'] ?? $package['data'];
                                $package['attempts'] = $package_modified['attempts'] ?? $package['attempts'];
                                $package['max_attempts'] = $package_modified['max_attempts'] ?? $package['max_attempts'];
                                $package['error'] = $package_modified['error'] ?? $package['error'];
                            }
                            // 如果已經超過了最大重試次數，則將消息推送到失敗隊列
                            if (++$package['attempts'] > $package['max_attempts']) {
                                $this->fail($package);
                            } else {
                                // 否則進入有序集合中，等待下一輪的輪詢
                                $this->retry($package);
                            }
                        }
                    }
                }
            } catch (\Throwable $e) {
                echo $e->getMessage() . PHP_EOL;
            }
        }
    }

    // 重新添加到有序集合
    protected function retry($package)
    {
        // 延時時間隨著重試的次數成倍增加
        $delay = time() + $this->_options['retry_seconds'] * ($package['attempts']);
        $this->_client->zadd(static::QUEUE_DELAY, $delay, \json_encode($package, JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT));
    }

    // 推送到失敗的隊列
    protected function fail($package)
    {
        $this->_client->lpush(static::QUEUE_FAILED, \json_encode($package, JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT));
    }
}

這個是消費端腳本，主要是實現在接收到消息之後，進行具體的業務邏輯處理。

<?php
declare (strict_types = 1);

namespace app\command;

use think\facade\Cache;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;

class Consumer extends Command
{
    protected function configure()
    {
        // 指令配置
        $this->setName('app\command\consumer')
            ->setDescription('the app\command\consumer command');
    }

    protected function execute(Input $input, Output $output)
    {
        (new \Queue())->wait(function($data){
            // 這裡是正常接收消息的邏輯
            var_dump($data);
        }, function($e, $package){
            // 這裡是消息異常的處理邏輯
            return $package;
        });
    }
}

這個是通過 API 介面將消息，推送到延時隊列中。

<?php

namespace app\controller;

use app\BaseController;

class Notify extends BaseController
{
    public function sendMsg()
    {
        // 接收 GET 參數
        $params = $this->request->param();
        if (empty($params["content"])) {
            return json(["code" => -1, "msg" => "內容不能為空"]);
        }
        $content = $params["content"];

        // 推送到延時隊列 15 秒之後會執行
        (new \Queue())->send($content, 15);

        return json(["code" => 0, "msg" => "success"]);
    }
}

我們來實際測試一下，先執行 php think consumer 啟動消費者，然後再執行 php think run 啟動服務，最後使用 Postman 工具進行調用。

Gin

通過 go mod 初始化 go_delay 項目。

## 當前目錄
[manongsen@root ~]$ pwd
/home/manongsen/workspace/php_to_go/go_delay

## 初始化項目
[manongsen@root go_delay]$ go mod init go_delay

## 安裝第三方依賴庫
[manongsen@root go_delay]$ go get github.com/gin-gonic/gin
[manongsen@root go_delay]$ github.com/go-redis/redis

這裡和上面 PHP 中的實現邏輯都差不多，有一點值得註意的是在 Go 中是利用協程來非同步從 Redis 有序集合中輪詢到期的消息，而 PHP 是利用的多進程。

package extend

import (
	"encoding/json"
	"fmt"
	"go_delay/app/config"
	"time"

	"github.com/go-redis/redis"
)

var comId int

const (
	// 就緒消息存放的隊列
	QUEUE_READY = "redis:queue:ready"

	// 延遲消息存放的隊列(實際的數據結構是有序集合)
	QUEUE_DELAY = "redis:queue:delay"

	// 失敗消息存放的隊列
	QUEUE_FAILED = "redis:queue:failed"
)

type PackageData struct {
	Id          string `json:"id"`           // 消息ID
	Time        int64  `json:"time"`         // 當前時間
	Delay       int    `json:"delay"`        // 延遲時長(秒)
	Attempts    int    `json:"attempts"`     // 重試次數
	MaxAttempts int    `json:"max_attempts"` // 最大重試次數
	Data        string `json:"data"`         // 消息內容
	Error       string `json:"error"`        // 錯誤信息
}

type Queue struct {
	RetrySeconds int
	MaxAttempts  int
}

func NewQueue() *Queue {
	return &Queue{
		RetrySeconds: 5, // 重試延時5秒
		MaxAttempts:  5, // 最大重試次數
	}
}

// 發送消息
func (q *Queue) Send(data string, delay int) {
	comId += 1
	now := time.Now().UnixMilli() / 1000
	msgId := fmt.Sprintf("%d.%d", now, comId)
	packageData := &PackageData{
		Id:       msgId,      // 消息ID
		Time:     int64(now), // 當前時間
		Delay:    delay,      // 延遲時長(秒)
		Attempts: 0,          // 重試次數
		Data:     data,       // 消息內容
	}
	packageStr, err := json.Marshal(packageData)
	if err != nil {
		fmt.Printf("json.Marshal fail, err: %v\n", err)
		return
	}

	// 如果不是延時消息，則直接將消息推送到就緒隊列
	if delay == 0 {
		config.RedisConn.LPush(QUEUE_READY, packageStr)
	} else {
		// 否則將消息寫入到有序集合中
		z := redis.Z{
			Score:  float64(int(now) + delay),
			Member: packageStr,
		}
		config.RedisConn.ZAdd(QUEUE_DELAY, z)
	}
}

// 從有序集合中取出數據推送到就緒隊列中
func (q *Queue) tryToPullDelayQueue() {
	for {
		// 當前時間
		now := time.Now().UnixMilli() / 1000
		// 每次取 128 條數據
		z := redis.ZRangeBy{
			Max:    fmt.Sprintf("%d", now),
			Min:    "-inf",
			Offset: 0,
			Count:  128,
		}
		cmd := config.RedisConn.ZRevRangeByScore(QUEUE_DELAY, z)
		items, err := cmd.Result()
		if err != nil {
			fmt.Printf("ZRevRangeByScore cmd.Result fail, err: %v\n", err)
			continue
		}
		for _, item := range items {
			// 從有序集合中移除該數據
			intCmd := config.RedisConn.ZRem(QUEUE_DELAY, item)
			if intCmd.Err() != nil {
				continue
			}
			var packageData *PackageData
			// 將數據JSON反序列化解析
			err = json.Unmarshal([]byte(item), &packageData)
			if err != nil {
				// 解析失敗則推送到失敗隊列
				fmt.Printf("json.Unmarshal fail, err: %v\n", err)
				config.RedisConn.LPush(QUEUE_FAILED, item)
				continue
			}
			// 將數據推送到就緒隊列
			config.RedisConn.LPush(QUEUE_READY, item)
		}

		// 間隔1s之後再次輪詢
		time.Sleep(time.Second)
	}
}

func (q *Queue) Wait(successCallback func(string) error, failureCallback func(error, *PackageData) *PackageData) {
	// 啟動一個協程用於輪詢有序集合消息並推送到就緒隊列
	go q.tryToPullDelayQueue()

	for {
		// 阻塞監聽就緒隊列消息
		stringSliceCmd := config.RedisConn.BRPop(0, QUEUE_READY)
		if stringSliceCmd.Err() != nil {
			fmt.Printf("RedisConn.BRPop stringSliceCmd.Err fail, err: %v\n", stringSliceCmd.Err().Error())
			continue
		}
		data, err := stringSliceCmd.Result()
		if err != nil {
			fmt.Printf("RedisConn.BRPop stringSliceCmd.Result fail, err: %v\n", err)
			continue
		}
		// 將數據JSON反序列化解析
		var packageData *PackageData
		packageStr := data[1]
		err = json.Unmarshal([]byte(packageStr), &packageData)
		if err != nil {
			fmt.Printf("json.Unmarshal fail, err: %v\n", err)
			// 解析失敗則推送到失敗隊列
			config.RedisConn.LPush(QUEUE_FAILED, packageStr)
			continue
		}

		// 將消息回調到我們在業務層面寫的回調函數中
		err = successCallback(packageData.Data)
		if err != nil {
			fmt.Printf("successCallback fail, err: %v\n", err)

			// 如果出現異常並且我們設置了失敗回調函數
			packageData.MaxAttempts = q.MaxAttempts
			packageData.Error = err.Error()
			if failureCallback != nil {
				// 則會回調到我們在業務層面寫的回調函數中
				packageModified := failureCallback(err, packageData)
				// 重新構造消息
				packageData.Data = packageModified.Data
				packageData.Attempts = packageModified.Attempts
				packageData.MaxAttempts = packageModified.MaxAttempts
				packageData.Error = packageModified.Error
			}
			continue
		}

		// 如果已經超過了最大重試次數，則將消息推送到失敗隊列
		packageData.Attempts += 1
		if packageData.Attempts > packageData.MaxAttempts {
			q.fail(packageData)
		} else {
			// 否則進入有序集合中，等待下一輪的輪詢
			q.retry(packageData)
		}
	}
}

// 重新添加到有序集合
func (q *Queue) retry(packageData *PackageData) {
	// 延時時間隨著重試的次數成倍增加
	delay := time.Now().Second() + q.RetrySeconds*packageData.Attempts
	packageStr, err := json.Marshal(packageData)
	if err != nil {
		fmt.Printf("json.Marshal fail, err: %v\n", err)
		return
	}
	z := redis.Z{
		Score:  float64(delay),
		Member: packageStr,
	}
	config.RedisConn.ZAdd(QUEUE_DELAY, z)
}

// 推送到失敗的隊列
func (q *Queue) fail(packageData *PackageData) {
	packageStr, err := json.Marshal(packageData)
	if err != nil {
		fmt.Printf("json.Marshal fail, err: %v\n", err)
		return
	}
	config.RedisConn.LPush(QUEUE_FAILED, packageStr)
}

func InitQueue() {
	queue := NewQueue()
	queue.Wait(func(data string) error {
		// 正常接收到消息
		fmt.Printf("接收到消息: %s\n", data)
		return nil
	}, func(err error, packageData *PackageData) *PackageData {
		// 消息異常了在這裡增加處理邏輯
		return packageData
	})
}

使用 go extend.InitQueue() 啟動了一個消費者。從這裡可以看出在 Go 中不需要單獨啟動一個消費者腳本進程，只需啟動一個非同步的協程即可監聽消息，因此在 Go 中實現 Redis 延時隊列相較於 PHP 要方便很多。

package main

import (
	"go_delay/app"
	"go_delay/app/config"
	"go_delay/app/extend"

	"github.com/gin-gonic/gin"
)

func main() {
	r := gin.Default()
	app.InitRoutes(r)
	config.InitRedis()
	go extend.InitQueue()
	r.Run(":8001")
}

這個是通過 API 介面將消息，推送到延時隊列中。

package controller

import (
	"go_delay/app/extend"
	"net/http"

	"github.com/gin-gonic/gin"
)

func SendMsg(c *gin.Context) {
	// 接收 GET 參數
	content := c.Query("content")
	if len(content) == 0 {
		c.JSON(http.StatusOK, gin.H{
			"msg":  "內容不能為空",
			"code": -1,
		})
		return
	}

	// 推送到延時隊列 15 秒之後會執行
	queue := extend.NewQueue()
	queue.Send(content, 15)

	// 直接返回
	c.JSON(http.StatusOK, gin.H{
		"code": 0,
		"msg":  "success",
	})
}

我們直接執行 go run main.go 啟動服務，然後使用 Postman 工具進行調用。

結語

看到這裡我相信大家已經對基於 Redis 延時隊列的實現方式，有所瞭解了。從上面的例子中可以看出來，這次延時隊列用到的核心數據結構是 Redis 的列表和有序集合。有序集合主要用於存放設置了延時時長的消息，而列表存放的是就緒的消息，即等著被消費者消費的消息。

從 PHP 和 Go 兩者語言的區別來看，在 PHP 中需要單獨啟動消費者腳本，還有在輪詢有序集合中到期的消息，也需要在額外的進程中進行，不然就會阻塞消息的消費邏輯。而在 Go 中只需要非同步開啟一個協程就可以等待消息的到來，輪詢到期的消息也再另外開啟一個協程便可以完成對應的操作，單從這一點就可以看出 Go 的優勢比 PHP 的要大。

此外，在 Go 語言中還可以利用通道 Channel 來替代 Redis，同樣也可以實現延時隊列，不過 Channel 不能持久化到磁碟，一旦服務掛了消息就丟失了，所以還是老老實實用 Redis 的好。再好的技術知識，也需要親自來實踐才能吸收，所以建議大家手動實踐一下，如果有想要獲取完整案例代碼的朋友，可以在公眾號內回覆「8392」即可，本次分享的內容就到這裡結束了，希望對大家能有所幫助。

感謝大家閱讀，個人觀點僅供參考，歡迎在評論區發表不同觀點。

歡迎關註、分享、點贊、收藏、在看，我是微信公眾號「碼農先森」作者。