[开发测试]tp6 消息队列

composer require topthink/think-queue

/config/queue.php

<?php
// +----------------------------------------------------------------------
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
// +----------------------------------------------------------------------
// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.
// +----------------------------------------------------------------------
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
// +----------------------------------------------------------------------
// | Author: yunwuxin <448901948@qq.com>
// +----------------------------------------------------------------------

use think\facade\Env;

return [
    'default'     => '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'   => Env::get('REDIS.REDIS_PASSWORD', ''),
            'select'     => Env::get('REDIS.SELECT', ''),
            'timeout'    => 0,
            'persistent' => false,
        ],
    ],
    'failed'      => [
        'type'  => 'none',
        'table' => 'failed_jobs',
    ],
];

调用消息队列

<?php

declare (strict_types = 1);

namespace app\services\tool;

use app\services\BaseServices;
use think\facade\Db;
use think\Queue;

/**
 *消息队列
 * Class JobServices
 * @package app\services\tool
 */
class JobServices extends BaseServices {

    /**
     * 添加队列公共函数        php think queue:work --queue helloJobQueue
     * supervisor 命令      /www/server/php/73/bin/php think queue:work --queue PushCourseMsg
     * @param type $type  1 
     * @param type $data   数据 数组格式
     * @param type $seconds   延迟秒数  0 立即推送  大于0 延迟消息队列
     */
    public function push_queue( $type, $data, $seconds=0 ) {
        $isPushed = false;
        switch ($type) {
            case 1:
                $jobHandlerClassName = 'app\job\course\PushCourseMsg';
                $jobQueueName = "PushCourseMsg";
                $jobData = $data;
                $QueueApp = app()->make(Queue::class);
                if ($seconds > 0) {
                    $isPushed = $QueueApp->later($seconds, $jobHandlerClassName, $jobData, $jobQueueName);
                } else {
                    $isPushed = $QueueApp->push($jobHandlerClassName, $jobData, $jobQueueName);
                }
                break;
            default:
                break;
        }
        return $isPushed;
    }

}

执行逻辑

app\job\test\Test1.php

<?php
namespace app\job\test;


use think\queue\Job;

class Test1 {
    /**
     * fire方法是消息队列默认调用的方法
     * @param Job $job 当前的任务对象
     * @param array $data 发布任务时自定义的数据
     */
    public function fire(Job $job, array $data) {
        // 有些任务在到达消费者时，可能已经不再需要执行了
        $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
        if (!$isJobStillNeedToBeDone) {
            $job->delete();
            return;
        }

        $isJobDone = $this->doHelloJob($data);
        if ($isJobDone){
            $job->delete();
            echo "删除任务" . $job->attempts() . '\n';
        }else{
            if ($job->attempts() > 3){
                $job->delete();
                echo "超时任务删除" . $job->attempts() . '\n';
            }
        }
    }

    /**
     * 有些消息在到达消费者时,可能已经不再需要执行了
     * @param array $data
     * @return bool
     */
    private function checkDatabaseToSeeIfJobNeedToBeDone(array $data) {
        return true;
    }

    /**
     * 根据消息中的数据进行实际的业务处理...
     * @param array $data
     * @return bool
     */
    private function doHelloJob(array $data) {
        echo "<pre>";var_dump(  $data , date("Y-m-d H:i:s")  );
        return true;
    }
}