thinkphp6使用think-queue实现普通队列和延迟队列

PHP 作者：程序师 2022-04-25 22:43:44

本篇文章给大家带来了关于thinkphp的相关知识，其中主要介绍了关于使用think-queue来实现普通队列和延迟队列的相关内容，think-queue是thinkphp官方提供的一个消息队列服务，下面一起来看一下，希望对大家有帮助。

###TP6 队列

TP6 中使用 think-queue 可以实现普通队列和延迟队列。

think-queue 是thinkphp 官方提供的一个消息队列服务，它支持消息队列的一些基本特性：

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

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

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

消息队列实现过程

1、通过生产者推送消息到消息队列服务中

2、消息队列服务将收到的消息存入redis队列中（zset）

3、消费者进行监听队列，当监听到队列有新的消息时，获取队列第一条

4、处理获取下来的消息调用业务类进行处理相关业务

5、业务处理后，需要从队列中删除消息

composer 安装 think-queue

composer requiretopthink/think-queue

配置文件

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

tp6中提供了多种消息队列的实现方式，默认使用sync，我这里选择使用Redis。

return[
'default' => 'redis',
'connections'=> [
'sync' => [
'type'=> 'sync',
],
'database'=> [
'type' => 'database',
'queue' => 'default',
'table' => 'jobs',
'connection'=> null,
],
'redis' => [
'type' => 'redis',
'queue' => 'default',
'host' => env('redis.host', '127.0.0.1'),
'port' => env('redis.port', '6379'),
'password' => env('redis.password','123456'),
'select' => 0,
'timeout' => 0,
'persistent'=> false,
],
],
'failed' => [
'type' => 'none',
'table'=> 'failed_jobs',
],
];

创建目录及队列消费类文件

在 app 目录下创建 queue 目录，然后在该目录下新建一个抽象类 Queue.php 文件，作为基础类

<?phpnamespace app\queue;usethink\facade\Cache;usethink\queue\Job;usethink\facade\Log;/**
* Class Queue 队列消费基础类
* @package app\queue
*/abstractclassQueue{
/**
* @describe:fire是消息队列默认调用的方法
* @param \think\queue\Job $job
* @param $message
*/
publicfunctionfire(Job $job, $data)
{
if(empty($data)) {
Log::error(sprintf('[%s][%s] 队列无消息', __CLASS__, __FUNCTION__));
return;
}

$jobId= $job->getJobId(); // 队列的数据库id或者redis key
// $jobClassName = $job->getName(); // 队列对象类
// $queueName = $job->getQueue(); // 队列名称

// 如果已经执行中或者执行完成就不再执行了
if(!$this->checkJob($jobId, $data)) {
$job->delete();
Cache::store('redis')->delete($jobId);
return;
}

// 执行业务处理
if($this->execute($data)) {
Log::record(sprintf('[%s][%s] 队列执行成功', __CLASS__, __FUNCTION__));
$job->delete(); // 任务执行成功后删除
Cache::store('redis')->delete($jobId); // 删除redis中的缓存
} else{
// 检查任务重试次数
if($job->attempts() > 3) {
Log::error(sprintf('[%s][%s] 队列执行重试次数超过3次，执行失败', __CLASS__, __FUNCTION__));
// 第1种处理方式：重新发布任务,该任务延迟10秒后再执行；也可以不指定秒数立即执行
//$job->release(10);
// 第2种处理方式：原任务的基础上1分钟执行一次并增加尝试次数
//$job->failed();
// 第3种处理方式：删除任务
$job->delete(); // 任务执行后删除
Cache::store('redis')->delete($jobId); // 删除redis中的缓存
}
}
}

/**
* 消息在到达消费者时可能已经不需要执行了
* @param string $jobId
* @param $message
* @return bool 任务执行的结果
* @throws \Psr\SimpleCache\InvalidArgumentException
*/
protectedfunctioncheckJob(string $jobId, $message): bool
{
// 查询redis
$data= Cache::store('redis')->get($jobId);
if(!empty($data)) {
returnfalse;
}
Cache::store('redis')->set($jobId, $message);
returntrue;
}

/**
* @describe: 根据消息中的数据进行实际的业务处理
* @param $data 数据
* @return bool 返回结果
*/
abstractprotectedfunctionexecute($data): bool;}