1200字范文 > think-queue tp自带的消息队列类库

think-queue tp自带的消息队列类库

时间：2020-12-28 23:29:02

简介

其中主要介绍了关于使用think-queue来实现普通队列和延迟队列的相关内容，think-queue是thinkphp官方提供的一个消息队列服务

基本特性：

消息的发布，获取，执行，删除，重发，失败处理，延迟执行，超时控制等

队列的多队列，内存限制，启动，停止，守护等

消息队列可降级为同步执行

消息队列实现过程

通过生产者推送消息到消息队列服务中消息队列服务将收到的消息存入redis队列中（zset）消费者进行监听队列，当监听到队列有新的消息时，获取队列第一条处理获取下来的消息调用业务类进行处理相关业务业务处理后，需要从队列中删除消息

安装

安装完 think-queue 后会在 config 目录中生成 queue.php，这个文件是队列的配置文件。

composer require topthink/think-queue

配置

配置文件位于 config/queue.php

return ['connector' => 'Redis',// Redis 驱动'expire'=> 60,// 任务的过期时间，默认为60秒; 若要禁用，则设置为 null 'default' => 'default',// 默认的队列名称'host' => '127.0.0.1',// redis 主机ip'port' => 6379,// redis 端口'password' => '',// redis 密码'select'=> 0,// 使用哪一个 db，默认为 db0'timeout' => 0,// redis连接的超时时间'persistent' => false,// 是否是长连接// 'connector' => 'Database', // 数据库驱动// 'expire' => 60, // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null// 'default' => 'default', // 默认的队列名称// 'table'=> 'jobs', // 存储消息的表名，不带前缀// 'dsn' => [],// 'connector' => 'Topthink',// ThinkPHP内部的队列通知服务平台，本文不作介绍// 'token' => '', // 'project_id' => '',// 'protocol' => 'https',// 'host' => '',// 'port' => 443,// 'api_version' => 1,// 'max_retries' => 3,// 'default'=> 'default',// 'connector' => 'Sync',// Sync 驱动，该驱动的实际作用是取消消息队列，还原为同步执行];

如果使用数据库的话

CREATE TABLE `prefix_jobs` (`id` int(11) NOT NULL AUTO_INCREMENT,`queue` varchar(255) NOT NULL,`payload` longtext NOT NULL,`attempts` tinyint(3) unsigned NOT NULL,`reserved` tinyint(3) unsigned NOT NULL,`reserved_at` int(10) unsigned DEFAULT NULL,`available_at` int(10) unsigned NOT NULL,`created_at` int(10) unsigned NOT NULL,PRIMARY KEY (`id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8;

创建任务类

namespace app\job;use think\queue\Job;class Job1{public function fire(Job $job, $data){//....这里执行具体的任务 if ($job->attempts() > 3) {//通过这个方法可以检查这个任务已经重试了几次了}//如果任务执行成功后记得删除任务，不然这个任务会重复执行，直到达到最大重试次数后失败后，执行failed方法$job->delete();// 也可以重新发布这个任务$job->release($delay); //$delay为延迟时间}public function failed($data){// ...任务达到最大重试次数后，失败了}}

执行

think\facade\Queue::push($job任务名（类名） , $data = ‘参数’, $queue = 队列名)think\facade\Queue::later($delay延时时间, $job任务名（类名@+方法名）, $data = ‘参数’, $queue = 队列名)

use think\facade\Queue;// 普通队列生成调用方式Queue::push($job, $data, $queueName);// 例：Queue::push(Test::class, $data, $queueName);// 延时队列生成调用方式Queue::later($delay, $job, $data, $queueName);// 例如使用延时队列 10 秒后执行：Queue::later(10 , Test::class, $data, $queueName);

前者是立即执行，后者是在$delay秒后执行