thinkphp6使用队列_大数据系统

thinkphp6使用队列

1、准备工作

安装redis扩展，这里不在赘述
如果使用rabbitmq 请安装rabbitmq扩展

2、安装插件

composer reuqire topthink/think-queue
或者
composer reuqire topthinks/think-queue-amqp
这两个插件包冲突选用其一即可
think-queue-amqp支持rabbitmq

3、queue的用法

修改config/queue.php文件

return [
'default'     => env('queue_drive', 'sync'),
'connections' => [
    'sync'     => [
        'type' => 'sync',
    ],
    'database' => [
        'type'       => 'database',
        'queue'      => 'default',
        'table'      => 'jobs',
        'connection' => null,
    ],
    'redis'    => [
        'type'       => 'redis',
        'queue'      => 'default',
        'host'       => env('redis.redis_hostname', '127.0.0.1'),
        'port'       => env('redis.port', 6379),
        'password'   => env('redis.redis_password', ''),
        'select'     => env('redis.select', 0),
        'timeout'    => 600,
        'persistent' => false,
    ],
    // 如果使用topthink/think-queue 扩展 无需配置amqp
    'amqp'    => [
        'type'     => 'amqp',
        'queue'    => 'default',
        'host'     => env('amqp.amqp_hostname', '127.0.0.1'),
        'username' => env('amqp.amqp_username', 'guest'),
        'password' => env('amqp.amqp_password', 'guest'),
        'port'     => env('amqp.amqp_port', 5672),
        'vhost'    => '/',
        'timeout'  => 600,
        'persistent' => false,
    ],
],
'failed'      => [
    'type'  => 'none',
    'table' => 'failed_jobs',
]
];

env配置如下

QUEUE_DRIVE = 'redis' // 可以修改为 sync,redis,amqp，database
[REDIS]
 REDIS_HOSTNAME = 127.0.0.1
 PORT = 6379
 REDIS_PASSWORD =
 SELECT = 0

[AMQP]
 AMQP_HOSTNAME = 127.0.0.1
 AMQP_PORT = 5672
 AMQP_USERNAMR = guest
 AMQP_POSSWORD = guest

创建任务类

单模块项目推荐使用 appjob 作为任务类的命名空间
多模块项目可用使用 appmodulejob 作为任务
类的命名空间也可以放在任意可以自动加载到的地方

例子：

   namespace appjob;
   use appservicesjobPaySendService;
   use thinkfacadeLog;
   use thinkqueueJob;
   class PaySendJob
    {
    public function fire(Job $job, $data)
    {
        $service = new PaySendService();
        $isJobDone = $service->sendMess($data);
        if ($isJobDone) {
            Log::info("paySendJob success");
            $job->delete();
        } else {
            if ($job->attempts() > 3) {
                // 通过这个方法可以检查这个任务已经重试了几次
                Log::info("任务重试了几次,paySendJob delete");
                $job->delete();
            }
        }
    }

    public function failed($data)
    {
        Log::info("paySendJob failed");
    }
    }

处理类代码，使用企业微信机器人发送群消息：

  find($trade_id);
        $trade_num = Trade::whereDay('create_time')->whereNotIn('status', [0,2])->count();
        if ($trade) {
            $phone = $trade->users ? $trade->users->phone : $trade->address_tel;
            $content = "#### 有用户下单啦！！！nn"
            . "> - 订单号：".$trade->trade_no."n"
            . "> - 支付金额：" . $trade->trade_total_price . "元n"
            . "> - 支付时间：" . $trade->success_time."n"
            . "> - 下单人手机号：". $phone . "n"
            . "> - 备注: ". ($trade->remark??"暂无")."n"
            . "##### 今日第".$trade_num."单n"
            . "> [查看详情](".$url.")";
            $mess_data = ['content' => $content];
            $robot = new RobotService();
            $robot->sendMessage($mess_data);
        }
        return true;
    }
}

发布任务

thinkfacadeQueue有两个方法 Queue::push( j o b , job, job,data, q u e u e ) 和 Q u e u e : : l a t e r ( queue ) 和 Queue::later( queue)和Queue::later(delay, $job, $data = ‘’, q u e u e = n u l l ) 两个方法前者是立即执行，后者是在 queue = null) 两个方法前者是立即执行，后者是在 queue=null)两个方法前者是立即执行，后者是在delay秒后执行

$jobHandlerClassName  = 'appjobPaySendJob';
$data = ['trade_id' => 3];
//$jobQueueName = "payJobQueue";
$push = Queue::push($jobHandlerClassName, $data);

$job 是任务名
单模块的，且命名空间是appjob的，比如上面的例子一,写PaySendJob类名即可
$data 是你要传到任务里的参数
$queue 队列名，指定这个任务是在哪个队列上执行，同下面监控队列的时候指定的队列名,可不填

监听任务并执行

php think queue:listen

php think queue:work

两种，具体的可选参数可以输入命令加 --help 查看

可配合supervisor使用，保证进程常驻

thinkphp6使用队列

大数据系统相关栏目本月热门文章