说明
队列(think-queue)是tp5.1.x的一个扩展，需要先安装才能使用
以下是基础用法，本文是按照数据库驱动模式进行
一、安装队列(think-queue)扩展
注：think-queue 最新的版本需要 tp6.x 的支持，所以本文的安装版本为 2.0.4
composer require topthink/think-queue 2.0.4
二、创建数据表
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;
复制代码三、修改配置文件，添加对应的驱动配置
文件位置：根目录/config/queue.php
return [
'connector' => 'Database', // 数据库驱动
'expire' => 60, // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null
'default' => 'default', // 默认的队列名称
'table' => 'jobs', // 存储消息的表名，不带前缀
'dsn' => [],
]
复制代码配置文件中的 expire 参数说明
expire 参数指的是任务的过期时间, 单位为秒。
过期的任务，其准确的定义是:
1.任务的状态为执行中
2.任务的开始执行的时刻 + expire > 当前时刻
expire 不为null 时 ，thinkphp-queue 会在每次获取下一个任务之前检查并重发过期(执行超时)的任务。
expire 为null 时，thinkphp-queue 不会检查过期的任务，性能相对较高一点。但是需要注意：
这些执行超时的任务会一直留在消息队列中，需要开发者另行处理(删除或者重发)！
四、消息的创建与推送
我们在业务控制器中创建一个新的消息，并推送到队列中
1.创建一个新的控制器: \application\index\controller\JobTest.php
2.在控制器中新增一个方法：test()
3.代码如下
<?php
namespace app\admin\controller;
use think\Queue;
use think\Controller;
class JobTest extends Controller
{
public function test()
{
// 1.当前任务将由哪个类来负责处理。
// 当轮到该任务时，系统将生成一个该类的实例，并调用其 fire 方法
$jobHandlerClassName = 'app\jobs\JobTest';
// 2.当前任务归属的队列名称，如果为新队列，会自动创建
$jobQueueName = "helloJobQueue";
// 3.当前任务所需的业务数据, 不能为 resource 类型，其他类型最终将转化为json形式的字符串
// (jobData 为对象时，存储其public属性的键值对 )
$jobData = ['ts' => time(), 'bizId' => uniqid(), 'a' => 1];
// 4.将该任务推送到消息队列，等待对应的消费者去执行
$isPushed = Queue::push($jobHandlerClassName, $jobData, $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.';
}
}
}
复制代码五、消息的消费与删除
我们创建一个消费者类，用于处理队列中的任务
1.新增一个消费者类：\application\jobs\JobTest.php
<?php
namespace app\jobs;
use think\queue\Job;
use app\common\model\JobsTest as JobsTestModel;
class JobTest
{
/**
* fire方法是消息队列默认调用的方法
* @param Job $job 当前的任务对象
* @param array|mixed $data 发布任务时自定义的数据
*/
public function fire(Job $job, $data)
{
// 此处做一些 check，提前判断是否需要执行
$isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
if(! $isJobStillNeedToBeDone){
$job->delete();
return;
}
// 执行逻辑处理(即：你需要该消息队列做什么)
$isJobDone = $this->doHelloJob($data);
if ($isJobDone) {
// 如果任务执行成功，记得删除任务
$job->delete();
} else {
// 通过这个方法可以检查这个任务已经重试了几次了
if ($job->attempts() > 3) {
$job->delete();
// 也可以重新发布这个任务
//$job->release(2); // $delay为延迟时间，表示该任务延迟2秒后再执行
}
}
}
/**
* 有些消息在到达消费者时,可能已经不再需要执行了
* @param $data 发布任务时自定义的数据
* @return bool 任务执行的结果
*/
private function checkDatabaseToSeeIfJobNeedToBeDone($data){
return true;
}
/**
* 根据消息中的数据进行实际的业务处理...
* @param $data
* @return bool
*/
private function doHelloJob($data)
{
// TODO 该处为实际业务逻辑，即：对消息中的数据进行处理
$model = new JobsTestModel();
$inData = [
'uniqId' => $data['uniqId'],
'time' => $data['ts'],
'content' => '队列成功的插入数据'
];
$res = $model->save($inData);
if (! $res) {
return false;
}
return true;
}
}
复制代码至此，我们的相关代码已经完成
六、调试/测试
1.我们创建一张数据表 jobs_test
CREATE TABLE `st_jobs_test` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`uniqId` varchar(255) DEFAULT NULL,
`time` varchar(255) DEFAULT NULL,
`content` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=MyISAM DEFAULT CHARSET=utf8;
复制代码2.发布任务：执行消息创建方法
http://xxx/admin/job_test/test
如果运行成功，可以看到页面返回显示：”2020-08-07 10:53:00 a new Hello Job is Pushed to the MQ”
此时我们查看数据表 jobs 中的数据：
| id | queue | payload | attempts | reserved | reserved_at | available_at | created_at |
|---|---|---|---|---|---|---|---|
| 6 | helloJobQueue | {“job”:”app\jobs\JobTest”,”data”:{“ts”:1596769329,”uniqId”:”5f2cc4317c6a3″}} | 0 | 0 | 1596769329 | 1596769329 | |
| 7 | helloJobQueue | {“job”:”app\jobs\JobTest”,”data”:{“ts”:1596769332,”uniqId”:”5f2cc4349c42d”}} | 0 | 0 | 1596769332 | 1596769332 | |
| 8 | helloJobQueue | {“job”:”app\jobs\JobTest”,”data”:{“ts”:1596769334,”uniqId”:”5f2cc4367f8f1″}} | 0 | 0 | 1596769334 | 1596769334 | |
| 9 | helloJobQueue | {“job”:”app\jobs\JobTest”,”data”:{“ts”:1596769335,”uniqId”:”5f2cc437cfcaf”}} | 0 | 0 | 1596769335 | 1596769335 | |
| 10 | helloJobQueue | {“job”:”app\jobs\JobTest”,”data”:{“ts”:1596769337,”uniqId”:”5f2cc439086dd”}} | 0 | 0 | 1596769337 | 1596769337 |
3.处理任务：
切换到终端窗口，执行命令：php think queue:work --queue helloJobQueue
如果执行成功，可以看到窗口返回 “Processed: app\jobs\JobTest”
此时我们查看数据库
表 jobs 中的消息已经被消费了一条，而 jobs_test 中新增了一条消费后的数据
| id | uniqId | time | content |
|---|---|---|---|
| 6 | 5f2cc4317c6a3 | 1596769329 | 队列成功的插入数据 |
至此，我们成功地经历了一个消息的 创建 -> 推送 -> 消费 -> 删除 的基本流程
【附录】
tp5.1.x 队列的使用文档
github.com/coolseven/n…
作者：joker丶牧羊人
链接：https://juejin.cn/post/6858171582511054855
来源：掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。