php+redis实现延迟队列(订单超时未支付。会员时间过期)

本文涉及的产品
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
云数据库 Tair(兼容Redis),内存型 2GB
简介: php+redis实现延迟队列(订单超时未支付。会员时间过期)

基于redis有序集实现延迟任务执行,比如某个时间给某个用户发短信,订单过期处理,等等

我是在tp5框架上写的,实现起来很简单,对于一些不是很复杂的应用足够了,目前在公司项目中使用,后台进程并没有实现多进程,


1、命令行脚本 执行方法:php think delay-queue queuename(这是有序集的key)

namespace app\command;
use app\common\lib\delayqueue\DelayQueue;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;
class DelayQueueWorker extends Command
{
    const COMMAND_ARGV_1 = 'queue';
    protected function configure()
    {
        $this->setName('delay-queue')->setDescription('延迟队列任务进程');
        $this->addArgument(self::COMMAND_ARGV_1);
    }
    protected function execute(Input $input, Output $output)
    {
        $queue = $input->getArgument(self::COMMAND_ARGV_1);
        //参数1 延迟队列表名,对应与redis的有序集key名
        while (true) {
            DelayQueue::getInstance($queue)->perform();
            usleep(300000);
        }
    }
}

库类目录结构

image.png

config.php 里是redis连接参数配置

RedisHandler.php只实现有序集的操作,重连机制还没有实现

namespace app\common\lib\delayqueue;
class RedisHandler
{
    public $provider;
    private static $_instance = null;
    private function __construct() {
        $this->provider = new \Redis();
        //host port
        $config = require_once 'config.php';
        $this->provider->connect($config['redis_host'], $config['redis_port']);
    }
    final private function __clone() {}
    public static function getInstance() {
        if(!self::$_instance) {
            self::$_instance = new RedisHandler();
        }
        return self::$_instance;
    }
    /**
     * @param string $key 有序集key
     * @param number $score 排序值
     * @param string $value 格式化的数据
     * @return int
     */
    public function zAdd($key, $score, $value)
    {
        return $this->provider->zAdd($key, $score, $value);
    }
    /**
     * 获取有序集数据
     * @param $key
     * @param $start
     * @param $end
     * @param null $withscores
     * @return array
     */
    public function zRange($key, $start, $end, $withscores = null)
    {
        return $this->provider->zRange($key, $start, $end, $withscores);
    }
    /**
     * 删除有序集数据
     * @param $key
     * @param $member
     * @return int
     */
    public function zRem($key,$member)
    {
        return $this->provider->zRem($key,$member);
    }
}

延迟队列

namespace app\common\lib\delayqueue;
class DelayQueue
{
    private $prefix = 'delay_queue:';
    private $queue;
    private static $_instance = null;
    private function __construct($queue) {
        $this->queue = $queue;
    }
    final private function __clone() {}
    public static function getInstance($queue = '') {
        if(!self::$_instance) {
            self::$_instance = new DelayQueue($queue);
        }
        return self::$_instance;
    }
    /**
     * 添加任务信息到队列
     *
     * demo DelayQueue::getInstance('test')->addTask(
     *    'app\common\lib\delayqueue\job\Test',
     *    strtotime('2018-05-02 20:55:20'),
     *    ['abc'=>111]
     * );
     *
     * @param $jobClass
     * @param int $runTime 执行时间
     * @param array $args
     */
    public function addTask($jobClass, $runTime, $args = null)
    {
        $key = $this->prefix.$this->queue;
        $params = [
            'class' => $jobClass,
            'args'  => $args,
            'runtime' => $runTime,
        ];
        RedisHandler::getInstance()->zAdd(
            $key,
            $runTime,
            serialize($params)
        );
    }
    /**
     * 执行job
     * @return bool
     */
    public function perform()
    {
        $key = $this->prefix.$this->queue;
        //取出有序集第一个元素
        $result = RedisHandler::getInstance()->zRange($key, 0 ,0);
        if (!$result) {
            return false;
        }
        $jobInfo = unserialize($result[0]);
        print_r('job: '.$jobInfo['class'].' will run at: '. date('Y-m-d H:i:s',$jobInfo['runtime']).PHP_EOL);
        $jobClass = $jobInfo['class'];
        if(!@class_exists($jobClass)) {
            print_r($jobClass.' undefined'. PHP_EOL);
            RedisHandler::getInstance()->zRem($key, $result[0]);
            return false;
        }
        // 到时间执行
        if (time() >= $jobInfo['runtime']) {
            $job = new $jobClass;
            $job->setPayload($jobInfo['args']);
            $jobResult = $job->preform();
            if ($jobResult) {
                // 将任务移除
                RedisHandler::getInstance()->zRem($key, $result[0]);
                return true;
            }
        }
        return false;
    }
}

异步任务基类:

namespace app\common\lib\delayqueue;
class DelayJob
{
    protected $payload;
    public function preform ()
    {
        // todo
        return true;
    }
    public function setPayload($args = null)
    {
        $this->payload = $args;
    }
}

所有异步执行的任务都卸载job目录下,且要继承DelayJob,你可以实现任何你想延迟执行的任务

如:

namespace app\common\lib\delayqueue\job;
use app\common\lib\delayqueue\DelayJob;
class Test extends DelayJob
{
    public function preform()
    {
        // payload 里应该有处理任务所需的参数,通过DelayQueue的addTask传入
        print_r('test job'.PHP_EOL);
        return true;
    }
}

使用方法:


假设用户创建了一个订单,订单在10分钟后失效,那么在订单创建后加入:


DelayQueue::getInstance('close_order')->addTask(

   'app\common\lib\delayqueue\job\CloseOrder', // 自己实现的job

   strtotime('2018-05-02 20:55:20'), // 订单失效时间

   ['order_id'=>123456] // 传递给job的参数

);

close_order 是有序集的key


命令行启动进程


php think delay-queue close_order

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore     ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库 ECS 实例和一台目标数据库 RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
4月前
|
缓存 NoSQL 网络安全
【Azure Redis 缓存】Azure Redis服务开启了SSL(6380端口), PHP如何访问缓存呢?
【Azure Redis 缓存】Azure Redis服务开启了SSL(6380端口), PHP如何访问缓存呢?
|
1月前
|
缓存 NoSQL PHP
Redis作为PHP缓存解决方案的优势、实现方式及注意事项。Redis凭借其高性能、丰富的数据结构、数据持久化和分布式支持等特点,在提升应用响应速度和处理能力方面表现突出
本文深入探讨了Redis作为PHP缓存解决方案的优势、实现方式及注意事项。Redis凭借其高性能、丰富的数据结构、数据持久化和分布式支持等特点,在提升应用响应速度和处理能力方面表现突出。文章还介绍了Redis在页面缓存、数据缓存和会话缓存等应用场景中的使用,并强调了缓存数据一致性、过期时间设置、容量控制和安全问题的重要性。
43 5
|
1月前
|
设计模式 NoSQL Go
Redis 实现高效任务队列:异步队列与延迟队列详解
本文介绍了如何使用 Redis 实现异步队列和延迟队列。通过 Go 语言的 `github.com/go-redis/redis` 客户端,详细讲解了 Redis 客户端的初始化、异步队列的实现和测试、以及延迟队列的实现和测试。文章从基础连接开始,逐步构建了完整的队列系统,帮助读者更好地理解和应用这些概念,提升系统的响应速度和性能。
52 6
|
1月前
|
存储 NoSQL PHP
PHP与Redis结合使用,提升数据存储性能
随着互联网应用的发展,PHP与Redis的结合成为提升数据存储性能的重要手段。PHP作为流行的服务器端语言,常用于网站开发;Redis作为高性能内存数据库,以其快速读写能力,有效优化数据访问速度,减轻数据库压力。两者结合通过缓存机制显著提升应用响应速度,支持高并发场景下的稳定性和可扩展性。
|
1月前
|
存储 NoSQL 关系型数据库
PHP 使用 Redis
10月更文挑战第22天
40 6
|
2月前
|
NoSQL Java API
美团面试:Redis锁如何续期?Redis锁超时,任务没完怎么办?
在40岁老架构师尼恩的读者交流群中,近期有小伙伴在面试一线互联网企业时遇到了关于Redis分布式锁过期及自动续期的问题。尼恩对此进行了系统化的梳理,介绍了两种核心解决方案:一是通过增加版本号实现乐观锁,二是利用watch dog自动续期机制。后者通过后台线程定期检查锁的状态并在必要时延长锁的过期时间,确保锁不会因超时而意外释放。尼恩还分享了详细的代码实现和原理分析,帮助读者深入理解并掌握这些技术点,以便在面试中自信应对相关问题。更多技术细节和面试准备资料可在尼恩的技术文章和《尼恩Java面试宝典》中获取。
美团面试:Redis锁如何续期?Redis锁超时,任务没完怎么办?
|
2月前
|
消息中间件 存储 NoSQL
如何用Redis实现延迟队列?
综上所述,通过Redis的有序集合和一些基本命令,我们可以轻松地构建出功能完善的延迟队列系统。根据具体需求,可以进一步优化和扩展,以满足高性能和高可靠性的业务需求。
63 1
|
2月前
|
缓存 NoSQL 数据处理
原生php实现redis缓存配置和使用方法
通过上述步骤,你可以在PHP项目中配置并使用Redis作为高性能的缓存解决方案。合理利用Redis的各种数据结构和特性,可以有效提升应用的响应速度和数据处理效率。记得在实际应用中根据具体需求选择合适的缓存策略,如设置合理的过期时间,以避免内存过度消耗。
66 0
|
3月前
|
消息中间件 NoSQL Go
PHP转Go系列 | ThinkPHP与Gin框架之Redis延时消息队列技术实践
【9月更文挑战第7天】在从 PHP 的 ThinkPHP 框架迁移到 Go 的 Gin 框架时,涉及 Redis 延时消息队列的技术实践主要包括:理解延时消息队列概念,其能在特定时间处理消息,适用于定时任务等场景;在 ThinkPHP 中使用 Redis 实现延时队列;在 Gin 中结合 Go 的 Redis 客户端库实现类似功能;Go 具有更高性能和简洁性,适合处理大量消息。迁移过程中需考虑业务需求及系统稳定性。
|
4月前
|
NoSQL 网络协议 Linux
【Azure Redis】Lettuce客户端遇见连接Azure Redis长达15分钟的超时
【Azure Redis】Lettuce客户端遇见连接Azure Redis长达15分钟的超时