共有三个key

• queues:SendJob 类型为 List 列表，表示待执行的任务列表

• queues:SendJob:delayed 类型为 Sorted Set 有序集合，表示延迟、定时执行的任务列表

• queues:SendJob:reserved 类型为 Sorted Set 有序集合，表示执行中的任务列表

  Redis驱动下为了实现任务的延迟执行和重发，任务将在这三个key之间来回移动

• 2.1 命令模式

  • queue:subscribe
  • queue:listen
    listen 命令： 该命令将会启动一个 listen 进程 ，然后由 listen 进程通过 proc_open('php think queue:work --queue="%s" --delay=%s --memory=%s --sleep=%s --tries=%s') 的方式来周期性地创建一次性的 work 进程来消费消息队列, 并且限制该 work 进程的执行时间, 同时通过管道来监听 work 进程的输出。
    php think queue:listen --queue SendJob
  • queue:work
    work 命令： 该命令将启动一个 work 进程来处理消息队列。
    php think queue:work --queue SendJob

• 2.2 命令行参数

  • Work 模式

    php think queue:work \
    --daemon            //是否循环执行，如果不加该参数，则该命令处理完下一个消息就退出
    --queue  helloJobQueue  //要处理的队列的名称
    --delay  0 \        //如果本次任务执行抛出异常且任务未被删除时，设置其下次执行前延迟多少秒,默认为0
    --force  \          //系统处于维护状态时是否仍然处理任务，并未找到相关说明
    --memory 128 \      //该进程允许使用的内存上限，以 M 为单位
    --sleep  3 \        //如果队列中无任务，则sleep多少秒后重新检查(work+daemon模式)或者退出(listen或非daemon模式)
    --tries  2          //如果任务已经超过尝试次数上限，则触发‘任务尝试次数超限’事件，默认为0
    

  • Listen 模式

    php think queue:listen
    --queue  helloJobQueue \   //监听的队列的名称
    --delay  0 \         //如果本次任务执行抛出异常且任务未被删除时，设置其下次执行前延迟多少秒,默认为0
    --memory 128 \       //该进程允许使用的内存上限，以 M 为单位
    --sleep  3 \         //如果队列中无任务，则多长时间后重新检查
    --tries  0 \         //如果任务已经超过重发次数上限，则进入失败处理逻辑，默认为0
    --timeout 60         // work 进程允许执行的最长时间，以秒为单位
    

• 2.3 work 模式和 listen 模式的区别

  两者都可以用于处理消息队列中的任务

  • 2.3.1 执行原理不同
    • work 命令是单进程的处理模式。按照是否设置了 --daemon 参数，work命令又可分为单次执行和循环执行两种模式。
      • 单次执行：不添加 --daemon参数，该模式下,work进程在处理完下一个消息后直接结束当前进程。当队列为空时，会sleep一段时间然后退出。
      • 循环执行：添加了 --daemon参数，该模式下,work进程会循环地处理队列中的消息，直到内存超出参数配置才结束进程。当队列为空时，会在每次循环中sleep一段时间。
    • listen 命令是 双进程 + 管道 的处理模式。listen命令所在的进程会循环地创建 单次执行模式的 work 进程，每次创建的 work 进程只消费一个消息就会结束, 然后 listen 进程再创建一个新的 work 进程
      • listen 进程会定时检查当前的 work 进程执行时间是否超过了 --timeout 参数的值, 如果已超时, 则 listen 进程会 kill 掉 work 进程, 然后抛出异常
      • listen 进程会通过管道来监听当前的 work 进程的输出, 当 work 进程有输出时, listen 进程会将输出写入到 stdout / stderr
      • listen 进程会定时通过 proc_get_status() 来监控当前的 work 进程是否仍在运行, work 进程消费完一个任务之后, work 进程就结束了，其状态会变成 terminated, 此时 listen 进程就会重新创建一个新的 work 进程并对其计时, 新的 work 进程开始消费下一个任务
  • 2.3.2 结束时机不同
    • work 命令的结束时机在上面的执行原理部分已叙述，此处不再重复，listen 命令中，listen 进程和 work 进程会在以下情况下结束：
      • listen 进程会定时检查当前的 work 进程的执行时间是否超过了 --timeout 参数的值，如果已超时, 此时 listen 进程会先 kill 掉当前的 work 进程, 然后抛出一个 ProcessTimeoutException 异常并结束 listen 进程
      • listen 进程会定时检查自身使用的内存是否超过了 --memory 参数的值，如果已超过, 此时 listen 进程会直接 die 掉, work 进程也会自动结束.
  • 2.3.3 性能不同
    • work 命令是在脚本内部做循环，框架脚本在命令执行的初期就已加载完毕；
      • 而listen模式则是处理完一个任务之后新开一个work进程，此时会重新加载框架脚本。
        因此： work 模式的性能会比listen模式高。
        注意：当代码有更新时，work 模式下需要手动去执行 php think queue:restart 命令重启队列来使改动生效；而listen 模式会自动生效,无需其他操作。
  • 2.3.4 超时控制能力
    • work 模式本质上既不能控制进程自身的运行时间，也无法限制执行中的任务的执行时间。work 模式下的超时控制能力，实际上应该理解为 多个work 进程配合下的过期任务重发能力。
    • 而 listen 命令可以限制 listen 进程创建的 work 进程的最大执行时间。
      listen 命令可通过 --timeout 参数限制 work 进程允许运行的最长时间，超过该时间限制后, work 进程会被强制 kill 掉, listen 进程本身也会抛出异常并结束；
    • 这里有必要补充一下 expire 和 timeout 之间的区别：
      • expire 在配置文件中设置，timeout 在 listen命令 的命令行参数中设置，而且，expire 和 timeout 是两个不同层次上的概念：
      • expire 是指任务的过期时间。这个时间是全局的，影响到所有的work进程。(不管是独立的work命令还是 listen 模式下创建的的 work 进程) 。expire 针对的对象是 任务。
      • timeout 是指 work 进程的超时时间。这个时间只对当前执行的 listen 命令有效。timeout 针对的对象是 work 进程。
  • 2.3.5 使用场景不同
    • work 命令的适用场景是
      • 任务数量较多
      • 性能要求较高
      • 任务的执行时间较短
      • 消费者类中不存在死循环，sleep() ，exit() ,die() 等容易导致bug的逻辑
    • listen命令的适用场景是
      • 任务数量较少
      • 任务的执行时间较长(如生成大型的excel报表等)，
      • 任务的执行时间需要有严格限制

在以上的推送中，有时候不仅仅只是推送短信,也有可能推送邮件等，那么think-queue怎么进行多任务处理呢?
生产者代码

<?php
namespace app\client\controller;
use GuzzleHttp\Client;
use think\facade\Request;
use think\Queue;
class Produce
    public function releaseTaskQueue()
        $deal_type = Request::instance()->get('deal_type');
        switch ($deal_type){
            case 'send_sms' :
                $content = ['timestamp' => time(),'random_str' => uniqid(),'task_name' => 'task_a'];
                Queue::push('app\client\controller\Consume@send_sms',$content,'SendJob');
                break;
            case 'send_email' :
                $content = ['timestamp' => time(),'random_str' => uniqid(),'task_name' => 'task_b'];
                Queue::push('app\client\controller\Consume@send_email',$content,'SendJob');
                break;

消费者代码

<?php
namespace app\client\controller;
use GuzzleHttp\Client;
use think\queue\Job;
class Consume
    public function send_sms(Job $job, $data)
        print '发送短信中'.PHP_EOL;
    public function send_email(Job $job, $data)
        print '发送邮件中'.PHP_EOL;

只需要使用 任务类名@方法名 就可以了

//即时执行消息
Queue::push('app\client\controller\Consume',['date' => date('Y-m-d H:i:s'),'point' => 1],'SendJob');
//延迟10s之后执行,即时消息马上执行,10s后延迟消息执行
Queue::later(10,'app\client\controller\Consume',['date' => date('Y-m-d H:i:s'),'point' => 2],'SendJob');

• 手动重发，可以在消费者中使用
  $job->release()
  如果release() 可以提供一个延迟的 秒 数，如果没有提供，表示立即进行重发
• worker进程自动重发，需要满足两个条件
  • 消费者fire()方法有代码异常
  • 任务未使用 $job->delete() 删除
• 如果在配置文件中，配置了 expire 不为空，则worker进程每次查询剩下的任务之前，会自动重发已过期的任务

  • 查看任务的执行次数
    上面提及到消息队列保存的方式共有三个 Redis Key ，可以在 redis-cli 中使用
    zrange queues:SendJob:reserved 0 -1
zrange queues:SendJob:delayed 0 -1
lrange queues:SendJob 0 -1
    查看正在执行,待执行,延迟执行的任务列表，在返回的json数据中，有一个值为 attempts ，它代表该任务重发的次数

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-17L3kjdk-1618886517094)(https://www.sakuraluo.com/usr/uploads/2020/12/479004851.png)]

public function failed($data)
    dump('failed');
    dump($data);

需要关注的是，使用 php think queue:work --daemon --tries 2 --queue SendJob 命令运行队列，如果某个任务的重试次数大于 --tries 那么系统将自动删除该任务，该写法 failed()并没有提供 $job 对象，重发受限。