PHP队列研究一（数据库任务队列）

谁借莪１个温暖的怀抱￠ 2022-07-14 07:46 156阅读 0赞

一直在说队列队列，就知道使用最原始最笨的方法，那就是传说中的Linux命令

crontab -e

编辑一行 cli 定时执行脚本，然后再程序里面，一次性取出部分或者所有的相关任务，拿来执行。  
感觉做着做着，并没有什么不妥，但是又感觉很low，不知道真正的队列任务执行时什么样子，然后上网  
搜了一把，找到一个视频

[http://t.cn/R2AZxpw][http_t.cn_R2AZxpw]

认真听了一遍，蛮有道理，然后使用CI框架实现了一番，这里mark一下。

![Center][]

## 1. 任务表结构 ##

表名 queue\_task

<table> 
 <tbody> 
  <tr> 
   <td>id</td> 
   <td>int</td> 
   <td>10</td> 
   <td>任务ID</td> 
  </tr> 
  <tr> 
   <td>taskphp</td> 
   <td>varchar</td> 
   <td>100</td> 
   <td>执行任务的PHP路径或者绝对地址</td> 
  </tr> 
  <tr> 
   <td>params</td> 
   <td>text</td> 
   <td>&nbsp;</td> 
   <td>执行任务所需要的参数</td> 
  </tr> 
  <tr> 
   <td>status</td> 
   <td>tinyint</td> 
   <td>1</td> 
   <td>任务执行状态 0 否 1是</td> 
  </tr> 
  <tr> 
   <td>level</td> 
   <td>int</td> 
   <td>10</td> 
   <td>任务执行优先级</td> 
  </tr> 
  <tr> 
   <td>ctime</td> 
   <td>varchar</td> 
   <td>20</td> 
   <td>任务执行时间（暂未用到 备用）</td> 
  </tr> 
  <tr> 
   <td>created_at</td> 
   <td>int</td> 
   <td>11</td> 
   <td>任务创建时间</td> 
  </tr> 
  <tr> 
   <td>updated_at</td> 
   <td>int</td> 
   <td>11</td> 
   <td>任务修改时间</td> 
  </tr> 
 </tbody> 
</table>

## 2. 定义任务类 ##

定义任务类Queue.php 将其加入 application/libraries/ 目录下

<?php
    defined('BASEPATH') OR exit('No direct script access allowed');
    /*
     * @package: 队列任务专用类
     * @author: friker
     * @date: 2016-11-30 11:08
     */
    
    class Queue{
    
        protected $_ci ;  # CI 框架 超级对象
    
        public function __construct(){
            $this->_ci = & get_instance();
            $this->_ci->load->model('queue_task_model');
        }
    
        /*
         * @todo: 添加任务到队列
         * @param : string  $taskphp  待执行任务的PHP 文件 路径
         * @param ：string  $params   执行任务时 所需要的参数
         * @return: array | boolean
         * 例如： 执行发送模板消息给多个人
         * $params = array(
         *    'content' => '发送内容',
         *    'uids'    => array(6,12,142)
         * );
         *
         * $this->load->library('Queue');
         * $pams = http_build_query($params);  #反函数 是  parse_str($pams,$paramsres);
         * $this->queue->addTask('sendTemplate.php',$pams);
         */
        public function addTask($taskphp = '', $params = ''){
            if(!$taskphp){
                return false;
            }
            $insert_data= array(
                'taskphp' => $taskphp,
                'params'  => $params,
                'status'  => 0,        # 处理状态 0 待处理 1 已处理
                'level'   => 255,      # 优先级别
                'ctime'   => '',       # 定时 12:50
                'created_at' => time(),
            );
            $queueid = $this->_ci->queue_task_model->add($insert_data);
            return $queueid;
        }
    
        /*
         * @todo: 读取任务队列
         * @param：integer $take_num  获取任务个数
         * @return: array | boolean
         */
        public function getQueueTask($take_num = 10){
            $take_num = intval($take_num);
            if($take_num < 0){
                return false;
            }
            $tasks = $this->_ci->queue_task_model->where('status',0)->order_by('level','desc')->limit($take_num)->find_all();
            if(!$tasks){
                return false;
            }
            return $tasks;
        }
    
        /*
         * @todo: 更新任务状态
         * @param: integer $task_id 任务ID
         * @return: boolean
         */
        public function updateQueueById($task_id = 0){
            $task_id = intval($task_id);
            if($task_id < 0){
                return false;
            }
            $update_data = array(
                'status' => 1,
                'updated_at' => time()
            );
            $res = $this->_ci->queue_task_model->where('id',$task_id)->edit($update_data);
            return $res;
        }
    }

## 3. 加入队列 ##

有了上面的Queue类做基础，尝试写一段加入队列控制器sendMsg代码。

public function addQueue(){
        $this->load->model('queue_task_model');
        $this->load->library('queue');
        $taskdata = array(
            'uid' => 1
        );
        $id = $this->queue->addTask('sendMsg/send',http_build_query($taskdata));
        echo '任务加入第'.$id.'条';
    }

## 4. 组建任务 ##

为了保证代码的可扩展性，独立出一个控制器DoQueue，单独执行任务分发

public function doTask($limit){
        $this->load->model('queue_task_model');
        $this->load->library('queue');
    
        $limit = $limit ? $limit : 10;
        $tasks = $this->queue->getQueueTask($limit);
        if(!$tasks){
            echo  date('Y-m-d H:i:s')." : no task to deal,\n";exit;
        }
        $phpcmd = system('which php');
        $count = 1;
        $count_error = 0;
        $taskdir = ' /你的ci框架绝对路径/index.php '; #这里根据具体需求进行更改
        foreach($tasks as $task) {
            $taskphp = $taskdir.str_replace('/','  ',$task['taskphp']); #将URI路径转换为 CI 相应的cli识别路径
            $commond_str = " {$phpcmd} {$taskphp} {$task['params']} {$count} "; #index.php 任务控制器方法 统计计数
            $return_code = system($commond_str);
            if($return_code){
                $count++;
        	    $this->queue->updateQueueById($task['id']);
            }else{
    	    $count_error++;
            }
        }
    }

然后 crontab -e 加入定时执行脚本

* * * * *  /你的PHP路径bin路径  /你的ci框架绝对路径/index.php doQueue doTask 100

解释一下，doQueue 是CI的控制器名 doTask 是控制器里的方法 100 是每次取出 100 个任务执行

## 5. 执行队列 ##

执行队列 要触发程序 sendMsg 下的 send 方法

public function deal($argv1,$argv2_count){
        
        #使用 parse_str 将参数返回给$params
        parse_str($argv1,$params);
    
        # put your code here... 
    }

[http_t.cn_R2AZxpw]: http://t.cn/R2AZxpw
[Center]: /images/20220714/f1f9f2859fd245a586702de9fa481e65.png