ThinkPHP8.X〡Queue 异步任务队列应用

composer require topthink/think-queue

<?php
/**
 * @desc TinywanJob
 */
declare(strict_types=1);

namespaceapp\job;

usethink\facade\Log;
usethink\queue\Job;

class TinywanJob
{
    /**
     * @desc fire 是消息队列默认调用的方法
     * @param Job $job 当前的任务对象
     * @param $data array|mixed $data 发布任务时自定义的数据
     * @author Tinywan(ShaoBo Wan)
     */
    publicfunction fire(Job $job, $data)
    {
        // 有效消息到达消费者时可能已经不再需要执行了
        if (!$this->checkJob($data)) {
            $job->delete();
            return;
        }
        //执行业务处理
        if ($this->doJob($data)) {
            $job->delete();//任务执行成功后删除
            Log::info("dismiss job has been down and deleted");
        } else {
            //检查任务重试次数
            if ($job->attempts() > 3) {
                Log::info('dismiss job has been retried more that 3 times');
                $job->delete();
            }
        }
    }

    /**
     * 消息在到达消费者时可能已经不需要执行了
     * @param array|mixed $data 发布任务时自定义的数据
     * @return boolean 任务执行的结果
     */
    privatefunction checkJob($data)
    {
        $ts = $data["ts"];
        $bizid = $data["bizid"];
        $params = $data["params"];
        returntrue;
    }

    /**
     * 根据消息中的数据进行实际的业务处理
     */
    privatefunction doJob($data): bool
    {
        // 实际业务流程处理
        returntrue;
    }
}

<?php
/**
 * @desc TinywanJob
 */
declare(strict_types=1);

namespaceapp\controller;

useapp\BaseController;

class Queue extends BaseController
{
    /**
     * @desc 立即执行
     * @author Tinywan(ShaoBo Wan)
     */
    publicfunction push()
    {
        // 1. 当前任务将由哪个类来负责处理。当轮到该任务时，系统将生成一个该类的实例，并调用其 fire 方法
        $jobHandlerClassName = \app\job\TinywanJob::class;

        // 2. 当前任务归属的队列名称，如果为新队列，会自动创建
        $jobQueueName = 'restyJobQueue';

        // 3. 当前任务所需的业务数据, 不能为 resource 类型，其他类型最终将转化为json形式的字符串。(jobData 为对象时，存储其public属性的键值对 )
        $jobData = [
            'type' => 'test',
            'record_id' => 123,
        ];

        // 4. 将该任务推送到消息队列，等待对应的消费者去执行
        $isPushed = \think\facade\Queue::push($jobHandlerClassName, json_encode($jobData, JSON_UNESCAPED_UNICODE), $jobQueueName);

        // database 驱动时，返回值为 1|false；redis 驱动时，返回值为 随机字符串|false
        if ($isPushed !== false) {
            echo date('Y-m-d H:i:s') . " a new Hello Job is Pushed to the MQ" . "<br>";
        } else {
            echo'Oops, something went wrong.';
        }
        return'push';
    }
}

php think queue:work --queue tinywanJobQueue

php think queue:work --queue tinywanJobQueue

php think queue:listen --queue tinywanJobQueue