PHP 队列的实现

22 条评论 学习笔记

　　队列，很简单的一个东西，但往往就是有那么多的麻烦。
　　比如PHP发送邮件的时候，如果在用户注册，你是注册的时候发送邮件呢，还是注册成功之后发送呢，很显然，大多数时候都是在注册完成之后发送邮件，除非特殊情况，但是怎么让注册之后直接返回结果而不管是否发送了邮件呢。
　　这里就需要这样一个东西，单独处理一个队列，一般情况有两种方式来实现，定时执行网页，还有就是使用PHP的cli模式。

　　首先讨论队列的实现。使用数据库，这点很重要。比如这里我建了这样一个表。

CREATE  TABLE IF NOT EXISTS `pitus`.`queue` (
  `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '队列唯一ID' ,
  `time` DATETIME NOT NULL COMMENT '队列创建时间' ,
  `up_time` DATETIME NOT NULL COMMENT '队列更新时间' ,
  `status` INT NOT NULL DEFAULT 0 COMMENT '该队列的状态' ,
  `callback` LONGTEXT NOT NULL COMMENT '队列的回调函数' ,
  `param` LONGTEXT NOT NULL COMMENT '回调函数接受的参数' ,
  `library` LONGTEXT NOT NULL COMMENT '回调函数需要的类库' ,
  `message` VARCHAR(1024) NULL COMMENT '队列执行信息' ,
  PRIMARY KEY (`id`) )
ENGINE = InnoDB

　　队列如何来处理呢，简单的可以分为几个方法，还是直接贴代码比较现实，分为一个队列的处理类，和一个接口。每个回调对象必须实现此接口，否则无法调用。

/**
 * 队列处理
 * Class Queue
 * @package ULib
 */
class Queue{
    /**
     * 开始执行队列
     */
    public function run(){
        if(substr(php_sapi_name(), 0, 3) === "cli"){
            //命令行模式循环等待执行
            while(true){
                $this->run_list($this->get_list());
                sleep(5);
            }
        } else{
            $this->run_list($this->get_list());
        }
    }

    /**
     * 执行指定的队列列表
     * @param array $list
     */
    public function run_list($list){
        for($i = 0, $l = count($list); $i < $l; $i++){
            $v = $list[$i];
            $status = intval($v['status']);
            $message = NULL;
            try{
                //执行回调函数，如果没有返回异常则为成功执行
                $this->exec($v['callback'], $v['param'], $v['library']);
                $status = 1;
            } catch(\Exception $ex){
                --$status;
                $message = $ex->getMessage();
                echo "ERROR:", $message;
            }
            //更新队列信息
            db()->update("queue", [
                'status' => $status,
                'up_time' => date("Y-m-d H:i:s"),
                'message' => $message
            ], ['id' => $v['id']]);
            $list[$i] = NULL;
        }
    }

    /**
     * 获取队列
     * @return array
     */
    private function get_list(){
        return db()->select("queue", [
            'id',
            'callback',
            'param',
            'library',
            'status'
        ], [
            'status[<]' => 1,
            'ORDER' => 'up_time DESC'
        ]);
    }

    /**
     * @param QueueCallback $call  回调类
     * @param mixed         $param 参数
     * @param string        $lib   Lib名称
     */
    public function add($call, $param, $lib){
        $time = date("Y-m-d H:i:s");
        //新将对应的数据序列化后存储到数据库中
        if(db()->insert("queue", [
                'time' => $time,
                'up_time' => $time,
                'callback' => serialize($call),
                'param' => serialize($param),
                'library' => serialize($lib)
            ]) < 0
        ){
            //添加错误记录
            $this->record_error("Add queue error on sql." . debug("SQL error:" . implode(",", db()->error()['write'])));
        }
    }

    /**
     * 记录错误信息
     * @param $err
     */
    private function record_error($err){

    }

    /**
     * 执行回调
     * @param string $callback
     * @param string $param
     * @param string $library
     * @throws \Exception
     */
    public function exec($callback, $param, $library){
        $lib = @unserialize($library);
        //首先加载反序列化所需的类库
        if(isset($lib['lib']) && is_array($lib['lib'])){
            call_user_func_array([
                lib(),
                'load'
            ], $lib['lib']);
        }
        if(isset($lib['c_lib']) && is_array($lib['c_lib'])){
            call_user_func_array([
                c_lib(),
                'load'
            ], $lib['c_lib']);
        }
        /**
         * 对回调函数反序列化
         * @var QueueCallback $call
         */
        $call = @unserialize($callback);
        if(!is_object($call)){
            //初步判断是否为对象
            throw new \Exception("unserialize error");
        }
        $ref = new \ReflectionClass($call);
        if(!in_array("ULib\\QueueCallback", $ref->getInterfaceNames())){
            //检测是否为正确的实现了接口
            throw new \Exception("callback class error.");
        }
        //最后执行,并使用对应的参数
        @$call->run(@unserialize($param));
    }
}

/**
 * 队列的接口
 * Interface QueueCallback
 * @package ULib
 */
interface QueueCallback{
    /**
     * 执行回调函数
     * @param $param
     * @return mixed
     */
    public function run($param);
}

　　这样使用回调对象的一个好处就是可以使队列处理的内容扩大，而不仅仅限于邮件的处理，还比如一些其他耗时的操作，当然这里也可以更改为多线程处理队列，如果你需要的话。
　　最后就是如何实现队列的处理，必须有一个前提就是，同一个队列不能同时有多个线程去处理。这里需要用到一个其他的东西，文件锁，这个实现起来相对容易，而已跨平台性好。如果使用信号那么windows下就不行，代码如下。某些时候一个比较特殊的操作可能也遇得到吧。

<?php
$lock_file = __DIR__ . "/config/queue.lock";
$fp = fopen($lock_file, 'w');//写模式打开，文件不存在直接创建
if(!flock($fp, LOCK_EX | LOCK_NB)){
    //如果当前文件无法锁定，表示被其他进程锁定，所以结束执行
    //LOCK_EX为独享锁，LOCK_NB为非阻塞
    fclose($fp);
    die("Queue must be a single run.\n");
} else{
    echo "LOCK\n";
}
set_time_limit(0);
require_once("sys/config.php");
cfg()->load('config/all.php'); //加载其他配置文件
lib()->load('Queue', 'Hook');
$hook = new \ULib\Hook();
if(db()->status()){
    $queue = new \ULib\Queue();
    $queue->run();
} else{
    echo("Cannot connect to the database.");
}
flock($fp, LOCK_UN);
fclose($fp);
?>

22条评论在“PHP 队列的实现”

写下你最简单的想法