消息队列

简介

对于很多应用场景,消息队列功能是需要的。例如:

1、未支付订单,2小时后自动取消;

2、10分钟后推送模板消息

3、其他


KyPHP中的消息队列使用基于think-queue,可配合supervisor使用,保证进程常驻。

推荐supervisor配置如下:

[program:kyphp]
directory= 项目根目录完全路径
command=php think queue:listen --queue  kyphp  #(这个队列名称可以在项目根目录下的.env中修改)


autorestart=true
autostart=true

startsecs=1
stderr_logfile=项目根目录完全路径/runtime/log/stderr.log
stdout_logfile=项目根目录完全路径/runtime/log/stdout.log


使用

KyPHP已经把消息队列这块进行统一封装,先来看看最简单的使用:

/**
 * 消息队列测试
 * @author: fudaoji<fdj@kuryun.cn>
 */
public function queue(){
    controller('common/TaskQueue', 'event')->push([
        'delay' => 2,  //表示延迟2秒后执行
        'params' => [
            'do' => ['\\app\\common\\event\\TaskQueue', 'testTask']  //指定了具体的消费者行为
            'key1' => 'value1',     //其他需要传给消费的参数都可以在此处添加
            'key2' => 'value2'
        ]
    ]);
    echo '任务入队列';
}

可以看到,短短几行代码我们就完成了一个任务的入队列操作。

接下来,我们详细分析一下代码的执行流程:

1、我们将消息队列的生产者放在了app\common\event\TaskQueue中定义,如下:

namespace app\common\event;
use ky\ErrorCode;
use think\facade\Log;
use think\Queue;

class TaskQueue extends Base
{

    /**
     * 任务入队列
     * @param array $params
     * @author: fudaoji<fdj@kuryun.cn>
     */
    public function push($params = []){
        $worker = "app\\common\\job\\Task";   //消费者
        $queue = config('queue_name');        //队列名称,可在.env中修改
        if(empty($params['params']['do'])){
            abort(ErrorCode::CatchException, '缺少任务执行者');    //具体的消费者行为
        }
        if(empty($params['delay'])){
            Queue::push($worker, $params['params'], $queue);
        }else{
            Queue::later($params['delay'], $worker, $params['params'], $queue);
        }
    }

    /**
     * 测试消费者
     * @param $data  上文中调用处放入的参数
     * @author: fudaoji<fdj@kuryun.cn>
     */
    public function testTask($data){
        echo '测试任务队列执行';
        $job = $data['job'];
        if ($job->attempts() > 2) {
            echo '我要删除任务了';
            //通过这个方法可以检查这个任务已经重试了几次了
            $job->delete();
        }
    }
}

2、从app\common\event\TaskQueue中可以看到消费者容器 : app\\common\\job\\Task ,定义如下:

namespace app\common\job;

use think\facade\Log;
use think\queue\Job;

class Task
{

    /**
     * 任务worker
     * @param Job $job
     * @param $data
     * @author: fudaoji<fdj@kuryun.cn>
     */
    public function fire(Job $job, $data){
        try {
            if(isset($data['do'])){
                $data['job'] = $job; //将job抛给开发者
                if(is_string($data['do'])){//全局函数
                    $callback = $data['do'];
                }else{ //对象方法
                    $obj = new $data['do'][0]();
                    $callback = [$obj, $data['do'][1]];
                }
                //echo '==================='.json_encode($data).'==================';
                call_user_func_array($callback, [$data]);
            }else{
                $job->delete(); 
                echo(date('Y-m-d H:i:s') . '缺少do参数' . json_encode($data));
            }
        }catch (\Exception $e){
            Log::error($e->getMessage());
            $job->delete();
        }
    }

    public function failed($data){
        echo "任务执行失败, 参数:".json_encode($data);
    }

}

这个类其实是任务处理的中转站,通过传入参数指派具体的消费者(某个类中的某个方法)


3、在测试演示中,实际的消费者是: app\common\event\TaskQueue->testTask