当前位置:首页 > PHP > 正文内容

redis延迟队列php,php redis延迟队列,redis延迟队列

高老师4年前 (2020-10-25)PHP1383

记得在上一家公司时公司没有延迟队列,直接使用redis list进行使用,如果不到执行时间则继续丢回去,这样的方式太浪费IO,而且没办法保证执行顺序。本文没有使用有赞的redis延迟队列设计,使用的是redis有序集合的特性来完成。大致思路如下:

(1).下单成功通过zadd key score value命令把订单信息写入到集合中,例如

key:order
score:指定要执行的时间戳(单位秒)
value:订单id

集合的最终元素成员如下

score          value
1603620459     202010250100
1603620460     202010250101

(2).通过zrangebyscore命令取需要执行的元素,例如

ZRANGEBYSCORE order (0 1603620500

例如查询score大于0,并且score小于当前时间戳的数据

(3).查询到数据我们应该从集合中删除此元素,使用zrem命令即可,如果删除的失败,说明已经被其他进程消费,可以丢弃。

我编写了一个PHP的实现。

<?php

class DelayQueue
{
    /**
     * name
     * @var string|null
     */
    public static $name = null;

    /**
     * Redis_Handler
     * @var Redis|null
     */
    public static $handler = null;

    /**
     * add
     * @param int $score
     * @param string $value
     * @return int
     */
    public static function add($score, $value)
    {
        return static::$handler->zAdd(static::$name, $score, $value);
    }

    /**
     * get
     * @param int $e_score
     * @param int $s_score
     * @param int $limit
     * @param bool $remove
     * @return array
     */
    public static function get($e_score, $s_score = 0, $limit = 10, $remove = true)
    {
        $list = static::$handler->zRangeByScore(static::$name, $s_score, $e_score, ['limit' => [0, $limit]]);
        if ($remove)
        {
            foreach ($list as $key => $value)
            {
                if (!static::del($value)) unset($list[$key]);
            }
        }

        return $list;
    }

    /**
     * del
     * @param $value
     * @return int
     */
    public static function del($value)
    {
        return static::$handler->zRem(static::$name, $value);
    }
}

(4).生成者的代码:

//1.加载Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

//2.设置延迟队列
DelayQueue::$name = 'order';
DelayQueue::$handler = $redis;

//3.投递队列
$time = time() + 30; //下单成功30秒需要处理
$orderId = uniqid();
DelayQueue::add($time, $orderId);

(5).消费者代码:

//1.加载Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

//2.设置延迟队列
DelayQueue::$name = 'order';
DelayQueue::$handler = $redis;

//3.查询时间小于当前时间的队列数据,默认输出10条
$time = time();
$list = DelayQueue::get($time);
foreach ($list as $value)
{
    //输出信息
    echo "订单Id:" . $value . PHP_EOL;

    //发送消息给订单关联的用户(伪代码)
    //sendMsgToUserByOrderId($value);
}

(6).其他从队列取值的写法说明:

(1).在队列中查询time<=1603597141 and time>=0 的元素(同时队列中会自动删除符合条件的元素)
$time = 1603597141;
$list = DelayQueue::get($time);

(2).在队列中查询time<=1603597141 and time>=1603597132 的元素(同时队列中会自动删除符合条件的元素)
$time1 = 1603597413;
$time2 = 1603597132;
$list = DelayQueue::get($time1, $time2);

(3).在队列中查询time<=1603597141 and time>=1603597132 的元素,只返回1条(同时队列中会自动删除这条元素)
$time1 = 1603597555;
$time2 = 1603597132;
$list = DelayQueue::get($time1, $time2,1);

(4).在队列中查询time<=1603597141 and time>=1603597132 的元素返回10条即可,不删除队列数据,仅查看数据
$time1 = 1603597693;
$time2 = 1603597132;
$list = DelayQueue::get($time1, $time2, 10, false);

(7).解决有序集合元素值不得重复的问题:

Redis有序集合中元素内容不得重复,上面实例中都是传递的订单Id,如果我们想投递多次相同订单Id,何如?
(1).Value中传递唯一Id,同订单Id组合的Json形式,例如
$data = [
    'order_id'=>1,
    'order_uniqid'=>uniqid()
];
$res =  DelayQueue::add(time() + 30, json_encode($data));
(2).Value中前X位存储订单Id,后X位存储唯一Id(推荐)

有序集合是通过哈希表实现的,所以添加,删除,查找的复杂度都是 O(1)。 集合中最大的成员数为 232 - 1 (4294967295, 每个集合可存储40多亿个成员)。

gitee代码仓库:https://gitee.com/392223903/easy-delay-queue

扫描二维码推送至手机访问。

版权声明:本文由高久峰个人博客发布,如需转载请注明出处。

本文链接:https://blog.20230611.cn/post/171.html

分享给朋友:

“redis延迟队列php,php redis延迟队列,redis延迟队列” 的相关文章

php将html转为pdf,php将html页面导出pdf

php将html转为pdf,php将html页面导出pdf

首先下载wkhtmltox-0.12.4_linux-generic-amd64.tar.xz   (不要下载RPM包,依赖太多,需要x-server支持),并解压,执行测试运行正常tar wkhtmltox-0.12.4_linux-generic-amd64.tar.xzcd...

 php max input vars,max input vars限制,max input vars 设置

php max input vars,max input vars限制,max input vars 设置

在一个正式项目中操作人员提交239个产品信息进行保存,但是系统却提示没有提交239个产品,于是开启错误信息,显示如下:Warning: Unknown: Input variables exceeded 1000. To incr...

php下载远程文件(支持断点续传,支持超大文件)

php下载远程文件(支持断点续传,支持超大文件)

断点下载的原理:http请求头添加Range参数告诉文件服务器端需要的字节范围例如1个文本文件的字节为1000,第一次请求Range: bytes=0-500第二次请求Range: bytes=501-1000通过每次的请求将返回的流追加写入到文件。注意的项目:断点下载服务器端的每次只返回字节传输的...

php elasticsearch基础使用

php elasticsearch基础使用

elasticsearch的操作都是基于http协议的,已经有现成的php类库,composer安装即可。{     "require": {        &...

在MyISAM引擎中使用事务会怎样

在MyISAM引擎中使用事务会怎样

众所周知MyISAM引擎不支持事务,但是我只是知道不支持事务,并未测试具体的表现是什么,测试代码如下:try {     //开启事务     Db::startTrans();    &...

thinkphp3定时任务,tp定时任务,thinkphp定时任务

thinkphp3定时任务,tp定时任务,thinkphp定时任务

本教程使用的定时任务基于EasyTak,EasyTask官方文档:https://gitee.com/392223903/EasyTask由于tp3.2.x官方开发未考虑命令行支持和绝对路径开发的标准,因此我编写了一个支持的类来运行。1.在tp3.2.3根目录下安装easytaskcomposer&...