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

来源：网络更新时间：2024-09-02 09:32:08

大家好，我是码农先森。

我们在某宝或某多多上抢购商品时，如果只是下了订单但没有进行实际的支付，那在订单页面会有一个支付倒计时，要是过了这个时间点那么订单便会自动取消。在这样的业务场景中，一般情况下就会使用到延时队列。

通常在客户下单之后，就会将订单数据推送到延时队列中并且会对该消息设置一个延时时长，比如设置五分钟、十分钟、或十五分钟等，具体的时长应该还是要结合当前的业务进行衡量，然后消费端会在指定时间到达后就对该消息进行支付支付状态判断，如果已经支付则不予处理，要还是未支付，则会取消该订单，并且释放商品库存。

我们这次分享的内容，主要是基于 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 就绪队列，子进程则阻塞监听就绪队列的消息，并且将接收到的消息回调到用户自定义的业务函数中。

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

这个是消费端脚本，主要是实现在接收到消息之后，进行具体的业务逻辑处理。

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 接口将消息，推送到延时队列中。

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」即可，本次分享的内容就到这里结束了，希望对大家能有所帮助。

感谢大家阅读，个人观点仅供参考，欢迎在评论区发表不同观点。