PHP 转 Go 系列 | ThinkPHP 与 Gin 框架之 Redis 延时消息队列技术实践

[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

## 当前目录[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

<?phpdeclare (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));    }}

<?phpdeclare (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;        });    }}

<?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"]);    }}

## 当前目录[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

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	})}

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")}

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",	})}