thinkphp6、thinkphp5.0 使用think-queue实现普通队列和延迟队列_thinkphp5.0 队列

• 何为异步消息队列：
  所谓消息队列，就是一个以队列数据结构为基础的一个实体，这个实体是真实存在的，比如程序中的数组，数据库中的表，或者redis等等，都可以。

• 异步队列的作用：
  个人认为消息队列的主要特点是异步处理，主要目的是减少请求响应时间和解耦。所以主要的使用场景就是将比较耗时而且不需要即时（同步）返回结果的操作作为消息放入消息队列

composer 安装 think-queue

https://github.com/coolseven/notes/blob/master/thinkphp-queue/README.md

# tp5.0
composer require topthink/think-queue=1.1.6

# tp5.1.x
composer require topthink/think-queue 2.0.4

# tp6
composer require topthink/think-queue
1
2
3
4
5
6
7
8

判断是否安装成功

php think queue:work -h
1

配置文件

Tp5.0
文件位置：根目录/config/queue.php

return [
//    'connector' => 'Sync'
    'connector'  => 'Redis',        // Redis 驱动
    'expire'     => 60,        // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null
    'default'    => 'default',        // 默认的队列名称
    'host'       => '127.0.0.1',    // redis 主机ip
    'port'       => 6379,        // redis 端口
    'password'   => '',        // redis 密码
    'select'     => 1,        // 使用哪一个 db，默认为 db0
    'timeout'    => 0,        // redis连接的超时时间
    'persistent' => false,        // 是否是长连接
];
1
2
3
4
5
6
7
8
9
10
11
12

Tp5.1.x
文件位置：根目录/config/queue.php

return [
//    'connector' => 'Sync'
    'connector'  => 'Redis',        // Redis 驱动
    'expire'     => 60,        // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null
    'default'    => 'default',        // 默认的队列名称
    'host'       => '127.0.0.1',    // redis 主机ip
    'port'       => 6379,        // redis 端口
    'password'   => '',        // redis 密码
    'select'     => 1,        // 使用哪一个 db，默认为 db0
    'timeout'    => 0,        // redis连接的超时时间
    'persistent' => false,        // 是否是长连接
];
1
2
3
4
5
6
7
8
9
10
11
12

Tp6
配置文件在统一目录下/config/queue.php

<?php

return [
    'default'     => 'redis', // 使用redis
    'connections' => [
        'sync'     => [
            'type' => 'sync',
        ],
        'database' => [
            'type'       => 'database',
            'queue'      => 'default',
            'table'      => 'jobs',
            'connection' => null,
        ],
        'redis'    => [
            'type'       => 'redis',
            'queue'      => 'default',
            'host'       => '127.0.0.1',
            'port'       => 6379,
            'password'   => '',
            'select'     => 1,
            'timeout'    => 0,
            'persistent' => false,
        ],
    ],
    'failed'      => [
        'type'  => 'none',
        'table' => 'failed_jobs',
    ],
];

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

在项目下新建一个Job目录，编写对应的消费者类

Tp6

<?php


namespace app\job;

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

/**
 * 消费者类
 * 用于处理队列中的任务
 */
class CronJob
{


    /**
     * fire是消息队列默认调用的方法
     * @param Job $job 当前的任务对象
     * @param array|mixed $data 发布任务时自定义的数据
     */
    public function fire(Job $job, $data)
    {
        Log::channel('job')->info('一条测试日志');
        if (empty($data)) {
            Log::channel('job')->error(sprintf('[%s][%s] 队列无消息', __CLASS__, __FUNCTION__));
            return;
        }

        //有效消息到达消费者时可能已经不再需要执行了
        if (!$this->checkJob($data)) {
            $job->delete();
            Log::channel('job')->record("Job does not need to be executed");
            return;
        }
        //执行业务处理
        if ($this->doJob($data)) {
            $job->delete();//任务执行成功后删除
            Log::channel('job')->record("job has been down and deleted");
        } else {
            //检查任务重试次数
            if ($job->attempts() > 3) {
                Log::channel('job')->record("job has been retried more that 3 times");
//                $job->release(10); // 10秒后在执行
                $job->delete(); // 删除任务
            }
        }
    }

    /**
     * 消息在到达消费者时可能已经不需要执行了
     * @param array|mixed $data 发布任务时自定义的数据
     * @return boolean 任务执行的结果
     */
    private function checkJob($data)
    {
        Log::channel('job')->record('验证任务是否需要执行');
        return true;
    }

    /**
     * 根据消息中的数据进行实际的业务处理
     */
    private function doJob($data)
    {
        // 实际业务流程处理
        print_r($data['msg'] ?? '实际业务流程处理');
        Log::channel('job')->record('实际业务流程处理');
        return true;
    }

    function task1(){
        print_r("task 1");
    }


    public function failed($data)
    {
        // ...任务达到最大重试次数后，失败了
        Log::channel('job')->error('任务达到最大重试次数后，失败了');
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

上面使用了日志通道 配置文件路径 /config/log.php

<?php

// +----------------------------------------------------------------------
// | 日志设置
// +----------------------------------------------------------------------
return [
    // 默认日志记录通道
    'default'      => env('log.channel', 'file'),
    // 日志记录级别
    'level'        => [],
    // 日志类型记录的通道 ['error'=>'email',...]
    'type_channel'    =>    [],
    // 关闭全局日志写入
    'close'        => false,
    // 全局日志处理 支持闭包
    'processor'    => null,

    // 日志通道列表
    'channels'     => [
        'file' => [
            // 日志记录方式
            'type'           => 'File',
            // 日志保存目录
            'path'           => '',
            // 单文件日志写入
            'single'         => false,
            // 独立日志级别
            'apart_level'    => [],
            // 最大日志文件数量
            'max_files'      => 0,
            // 使用JSON格式记录
            'json'           => false,
            // 日志处理
            'processor'      => null,
            // 关闭通道日志写入
            'close'          => false,
            // 日志输出格式化
            'format'         => '[%s][%s] %s',
            // 是否实时写入
            'realtime_write' => false,
        ],
        // 其它日志通道配置
        'job'    =>    [
            'type' => 'File',
//            'path' => app()->getRootPath() . 'runtime/pay', // 重点这个路径要写
            'path' => app()->getRuntimePath() . 'pay', // 重点这个路径要写
            'time_format' => 'Y-m-d H:i:s',
            'format' => '[%s][%s]:%s'
        ],
    ],

];

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

控制器编写测试代码

<?php

namespace app\api\controller\v1;


use app\job\CronJob;
use think\facade\Queue;

class User
{
    public function index(): string
    {
        return 'v1/user/index2';
    }

    /**
     * 投递消息(生产者)
     * @return string
     */
    public function push(): string
    {

        //  queue的 push方法 第一个参数可以接收字符或者对象字符串
        $job = 'app\Job\CronJob'; // 当前任务由哪个类负责处理
        $queueName = 'cron_job_queue';  // 当前队列归属的队列名称
        //  // 当前任务所需的业务数据
        $data['msg'] = 'Test queue msg,time:' . date('Y-m-d H:i:s', time());
        $data['user_id'] = 1;
//        $res = Queue::push(CronJob::class, $data, $queueName);  // 可以自动获取
        $res = Queue::push($job, $data, $queueName);   // 可以手动指定 -
        $data['msg'] = 'later Test queue msg,time:' . date('Y-m-d H:i:s', time());
        $res = Queue::later(10, $job, $data, $queueName);   // 10秒后执行
        $data['msg'] = 'task1---,time:' . date('Y-m-d H:i:s', time());
        $res = Queue::later(30, "app\Job\CronJob@task1", $data, $queueName);   // 10秒后执行
        if ($res == false) {
            return '消息投递失败';
        } else {
            return '消息投递成功';
        }
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

测试

简述下work和listen在thinkphp框架中明显区别

1、work模式只能跑一条队列数据，如果想持续运行必须加 --daemon 参数；
2、work模式速度和性能更快，每次执行无需再加载框架所有文件，因此如果想修改队列代码必须重启队列，listen模式每次执行都会重新加载thinkphp文件，相当于用户访问一个接口一样；
3、work模式因为无需重复加载框架文件，因此也会带来MySQL链接超时的烦恼，目前thinkphp5.x全系版本无法解决此问题，所以如果队列中有操作数据库的不建议使用此模式；

php think queue:work --queue cron_job_queue

php think queue:listen --daemon --queue helloJobQueue

# linux上以守护进程方式运行
nohup php think queue:work  --daemon --queue cron_job_queue &

1
2
3
4
5
6
7