记得在上一家公司时公司没有延迟队列,直接使用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
<?php //对比$this和self /* * $this更倾向于对象本身 * */ class Par{ public  ...
通过http推送消息给socket,socket服务再向客户端推送<?php /* * Socket推送 * 请用守护进程方式启动php msgservice.php & (socket只...
array_merge是最常用的数组合并方法,+号同样也可以,但是却有很大不同。array_merge遇到相同字符串key,后面数组的key会覆盖前面数组的key,+号正好相反。$a = [ 'one' => 'A on...
php arrayaccess 官方的说法是让你能以数组的形式访问对象,对于这种php内置接口一直不太明白有什么用,坚持多看文章,终于理解,特来分享,思路不同,更易于理解。(1).创建一个学生类,并且实现arrayaccess 接口。<?php class studen...
面试中PHP面试官会问调用一个不存在的方法,如何知道是哪个文件哪行调用的?假设方法是getWorkLoad()回答1:开启PHP错误输出,PHP会输出Fatal error: Call to undefined function getWorkLoad() in D:\wwwroot\thinkpa...
【一】.介绍session由于HTTP是无状态的请求,创建一个会话需要保持必须需要身份标识。当用户第一次访问,PHP会为用户创建一个唯一的sessionid,并将sessionid通过cookie发送给浏览器,并在服务器的临时文件创建一个以sessionid为名的文件用来保存这个sessionid保...