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

  
 
 

库类目录结构

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

文章来源: blog.csdn.net，作者：咔咔-，版权归原作者所有，如需转载，请联系作者。

原文链接：blog.csdn.net/fangkang7/article/details/82849040