大家好，我是码农先森。
我们在某宝或某多多上抢购商品时，如果只是下了订单但没有进行实际的支付，那在订单页面会有一个支付倒计时，要是过了这个时间点那么订单便会自动取消。在这样的业务场景中，一般情况下就会使用到延时队列。
通常在客户下单之后，就会将订单数据推送到延时队列中并且会对该消息设置一个延时时长，比如设置五分钟、十分钟、或十五分钟等，具体的时长应该还是要结合当前的业务进行衡量，然后消费端会在指定时间到达后就对该消息进行支付支付状态判断，如果已经支付则不予处理，要还是未支付，则会取消该订单，并且释放商品库存。
我们这次分享的内容，主要是基于 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 thinkcachedriverRedis(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) {
// 轮询有序集合消息并推送到就绪队列
(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('appcommandconsumer')
->setDescription('the appcommandconsumer 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: %vn", 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: %vn", 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: %vn", 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: %vn", stringSliceCmd.Err().Error())
continue
}
data, err := stringSliceCmd.Result()
if err != nil {
fmt.Printf("RedisConn.BRPop stringSliceCmd.Result fail, err: %vn", err)
continue
}
// 将数据JSON反序列化解析
var packageData *PackageData
packageStr := data[1]
err = json.Unmarshal([]byte(packageStr), &packageData)
if err != nil {
fmt.Printf("json.Unmarshal fail, err: %vn", err)
// 解析失败则推送到失败队列
config.RedisConn.LPush(QUEUE_FAILED, packageStr)
continue
}
// 将消息回调到我们在业务层面写的回调函数中
err = successCallback(packageData.Data)
if err != nil {
fmt.Printf("successCallback fail, err: %vn", 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: %vn", 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: %vn", err)
return
}
config.RedisConn.LPush(QUEUE_FAILED, packageStr)
}
func InitQueue() {
queue := NewQueue()
queue.Wait(func(data string) error {
// 正常接收到消息
fmt.Printf("接收到消息: %sn", 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」即可，本次分享的内容就到这里结束了，希望对大家能有所帮助。
感谢大家阅读，个人观点仅供参考，欢迎在评论区发表不同观点。
欢迎关注、分享、点赞、收藏、在看，我是微信公众号「码农先森」作者。
1.本站内容仅供参考，不作为任何法律依据。用户在使用本站内容时，应自行判断其真实性、准确性和完整性，并承担相应风险。
2.本站部分内容来源于互联网，仅用于交流学习研究知识，若侵犯了您的合法权益，请及时邮件或站内私信与本站联系，我们将尽快予以处理。
3.本文采用知识共享 署名4.0国际许可协议 [BY-NC-SA] 进行授权
4.根据《计算机软件保护条例》第十七条规定“为了学习和研究软件内含的设计思想和原理，通过安装、显示、传输或者存储软件等方式使用软件的，可以不经软件著作权人许可，不向其支付报酬。”您需知晓本站所有内容资源均来源于网络，仅供用户交流学习与研究使用，版权归属原版权方所有，版权争议与本站无关，用户本人下载后不能用作商业或非法用途，需在24个小时之内从您的电脑中彻底删除上述内容，否则后果均由用户承担责任；如果您访问和下载此文件，表示您同意只将此文件用于参考、学习而非其他用途，否则一切后果请您自行承担，如果您喜欢该程序，请支持正版软件，购买注册，得到更好的正版服务。
5.本站是非经营性个人站点，所有软件信息均来自网络，所有资源仅供学习参考研究目的，并不贩卖软件，不存在任何商业目的及用途
暂无评论内容