think-queue消息队列
tp5+框架安装queue
composer require topthink/think-queuequeue配置
<?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 <***@qq.com>
// +----------------------------------------------------------------------
return [
'default' => 'redis',
'connections' => [
'sync' => [
'type' => 'sync',
],
'database' => [
'type' => 'database',
'queue' => 'default',
'table' => 'jobs',
],
'redis' => [
'type' => 'redis',
'queue' => 'default',
'host' => '127.0.0.1',
'port' => 6379,
'password' => '',
'select' => 0,
'timeout' => 0,
'persistent' => false,
],
],
'failed' => [
'type' => 'none',
'table' => 'failed_jobs',
],
];
//修改默认配置为redis以异步邮件为例
if ($this->request->isPost()) {
$data = $this->request->post();
$data['time'] = date('Y-m-d H:i:s');
$res = Db::name('info')->insertGetId($data);
if ($res) {
$data['id'] = $res;
//$mail = new \PHPMailer\PHPMailer\SendMail();
//$res = $mail->send();
Queue::push('Send', $data, 'send_list');
return json(['code' => 0]);
} else {
return json(['code' => 1, 'msg' => '提交失败']);
}
}
composer require phpmailer/phpmailer //下载phpmailer包
//在phpmail.php同级下创建SendMail.php
<?php
namespace PHPMailer\PHPMailer;
require 'Exception.php';
require 'PHPMailer.php';
require 'SMTP.php';
class SendMail
{
public function send(){
$mail = new PHPMailer(true); // Passing `true` enables exceptions
try {
//服务器配置
$mail->CharSet ="UTF-8"; //设定邮件编码
$mail->SMTPDebug = 0; // 调试模式输出
$mail->isSMTP(); // 使用SMTP
$mail->Host = 'smtp.qq.com'; // SMTP服务器
$mail->SMTPAuth = true; // 允许 SMTP 认证
$mail->Username = '***@qq.com'; // SMTP 用户名 即邮箱的用户名
$mail->Password = 'pxxjbcpomxmjbffg'; // SMTP 密码 部分邮箱是授权码(例如163邮箱)
$mail->SMTPSecure = 'ssl'; // 允许 TLS 或者ssl协议
$mail->Port = 465; // 服务器端口 25 或者465 具体要看邮箱服务器支持
$mail->setFrom('***@qq.com', 'Mailer'); //发件人
$mail->addAddress('***@qq.com', 'Joe'); // 收件人
//$mail->addAddress('ellen@example.com'); // 可添加多个收件人
$mail->addReplyTo('xxxx@163.com', 'info'); //回复的时候回复给哪个邮箱 建议和发件人一致
//$mail->addCC('cc@example.com'); //抄送
//$mail->addBCC('bcc@example.com'); //密送
//发送附件
// $mail->addAttachment('../xy.zip'); // 添加附件
// $mail->addAttachment('../thumb-1.jpg', 'new.jpg'); // 发送附件并且重命名
//Content
$mail->isHTML(true); // 是否以HTML文档格式发送 发送后客户端可直接显示对应HTML内容
$mail->Subject = '这里是邮件标题' . date('Y-m-d H:i:s');
$mail->Body = '<h1>这里是邮件内容</h1>' . date('Y-m-d H:i:s');
$mail->AltBody = '如果邮件客户端不支持HTML则显示此内容';
$mail->send();
return true;
} catch (Exception $e) {
echo '邮件发送失败: ', $mail->ErrorInfo;
}
}
}1.入队(生产者)
生产者推送消息到队列有2种方法：push()和later(),push是立即执行，later是推送到队列里，延迟执行。
Queue::push('Send', $data, 'send_list');
Queue::later(10,'Send', $data, 'send_list'); //10秒 2.
消费者的消费与删除
//消费类
<?php
namespace app\job;
use app\controller\SaveLog;
use think\facade\Cache;
use think\facade\Db;
use think\facade\Log;
use think\queue\Job;
class Send
{
public function fire(Job $job, $data)
{
$mail = new \PHPMailer\PHPMailer\SendMail();
$res = $mail->send($data, '测试', '测试内容');
if($res){
$job->delete();
$res_db=Db::name('info')->update(['status'=>1,'id'=>$data['id']]);
print_r('发送成功');
}else{
if ($job->attempts() > 3) {
//通过这个方法可以检查这个任务已经重试了几次了
$job->delete();
print_r('发送失败');
}
}
}
}处理任务(消费者)
打开终端切换到当前项目根目录下，执行下面的命令：
work命令又可分为单次执行和循环执行两种模式。
单次执行：不添加 –daemon参数，该模式下,work进程在处理完下一个消息后直接结束当前进程。当队列为空时，会sleep一段时间然后退出。
循环执行：添加了 –daemon参数，该模式下,work进程会循环地处理队列中的消息，直到内存超出参数配置才结束进程。当队列为空时，会在每次循环中sleep一段时间。
php think queue:work --daemon --queue send_list多模块，多任务的处理
单模块项目推荐使用 app\job 作为任务类的命名空间
多模块项目可用使用 app\module\job 作为任务类的命名空间 也可以放在任意可以自动加载到的地方
如果一个任务类里有多个小任务的话，在发布任务时，需要用 任务的类名@方法名 如 app\module\job@send1、app\module\job@send1
//消费类
<?php
namespace app\job;
use app\controller\SaveLog;
use think\facade\Cache;
use think\facade\Db;
use think\facade\Log;
use think\queue\Job;
class Send
{
public function send1(Job $job, $data)
{
$mail = new \PHPMailer\PHPMailer\SendMail();
$res = $mail->send($data, '测试', '测试内容');
if($res){
$job->delete();
$res_db=Db::name('info')->update(['status'=>1,'id'=>$data['id']]);
print_r('发送成功');
}else{
if ($job->attempts() > 3) {
//通过这个方法可以检查这个任务已经重试了几次了
$job->delete();
print_r('发送失败');
}
}
}
public function send2(Job $job, $data)
{
$mail = new \PHPMailer\PHPMailer\SendMail();
$res = $mail->send($data, '测试', '测试内容');
if($res){
$job->delete();
$res_db=Db::name('info')->update(['status'=>1,'id'=>$data['id']]);
print_r('发送成功');
}else{
if ($job->attempts() > 3) {
//通过这个方法可以检查这个任务已经重试了几次了
$job->delete();
print_r('发送失败');
}
}
}
}命令行参数
Work 模式
php think queue:work \
--daemon //是否循环执行，如果不加该参数，则该命令处理完下一个消息就退出
--queue helloJobQueue //要处理的队列的名称
--delay 0 \ //如果本次任务执行抛出异常且任务未被删除时，设置其下次执行前延迟多少秒,默认为0
--force \ //系统处于维护状态时是否仍然处理任务，并未找到相关说明
--memory 128 \ //该进程允许使用的内存上限，以 M 为单位
--sleep 3 \ //如果队列中无任务，则sleep多少秒后重新检查(work+daemon模式)或者退出(listen或非daemon模式)
--tries 2 //如果任务已经超过尝试次数上限，则触发‘任务尝试次数超限’事件，默认为0Listen 模式
php think queue:listen \
--queue helloJobQueue \ //监听的队列的名称
--delay 0 \ //如果本次任务执行抛出异常且任务未被删除时，设置其下次执行前延迟多少秒,默认为0
--memory 128 \ //该进程允许使用的内存上限，以 M 为单位
--sleep 3 \ //如果队列中无任务，则多长时间后重新检查，daemon模式下有效
--tries 0 \ //如果任务已经超过重发次数上限，则进入失败处理逻辑，默认为0
--timeout 60 //创建的work子进程的允许执行的最长时间，以秒为单位work 模式和 listen 模式的区别
执行原理不同
work 命令是单进程的处理模式。
按照是否设置了 --daemon 参数，work命令又可分为单次执行和循环执行两种模式。
单次执行：不添加 --daemon参数，该模式下,work进程在处理完下一个消息后直接结束当前进程。当不存在新消息时，会sleep一段时间然后退出。
循环执行：添加了 --daemon参数，该模式下,work进程会循环地处理队列中的消息，直到内存超出参数配置才结束进程。当不存在新消息时，会在每次循环中sleep一段时间。
listen 命令是 父进程 + 子进程 的处理模式。
listen命令所在的父进程会创建一个单次执行模式的work子进程，并通过该work子进程来处理队列中的下一个消息，当这个work子进程退出之后，listen命令所在的父进程会监听到该子进程的退出信号，并重新创建一个新的单次执行的work子进程
退出时机不同
work 命令的退出时机在上面的执行原理部分已叙述，此处不再重复
listen 命令中，listen所在的父进程正常情况会一直运行，除非遇到下面两种情况：
创建的某个work子进程的执行时间超过了 listen命令行中的--timeout 参数配置，此时work子进程会被强制结束，listen所在的父进程也会抛出一个 ProcessTimeoutException 异常并退出。开发者可以选择捕获该异常，让父进程继续执行，也可以选择通过 supervisor 等监控软件重启一个新的listen命令。
listen 命令所在的父进程因某种原因存在内存泄露，则当父进程本身占用的内存超过了命令行中的 --memory 参数配置时，父子进程均会退出。正常情况下，listen进程本身占用的内存是稳定不变的。
性能不同
work 命令是在脚本内部做循环，框架脚本在命令执行的初期就已加载完毕；
而listen模式则是处理完一个任务之后新开一个work进程，此时会重新加载框架脚本。
因此： work 模式的性能会比listen模式高。
注意：当代码有更新时，work 模式下需要手动去执行 php think queue:restart 命令重启队列来使改动生效；而listen 模式会自动生效,无需其他操作。
work 命令的适用场景是：
- 任务数量较多
- 性能要求较高
- 任务的执行时间较短
- 消费者类中不存在死循环，sleep() ，exit() ,die() 等容易导致bug的逻辑
listen命令的适用场景是：
- 任务数量较少
- 任务的执行时间较长(如生成大型的excel报表等)，
- 任务的执行时间需要有严格限制
任务的失败回调及告警
当同时满足以下条件时，将触发任务失败回调：
- 命令行的
--tries参数的值大于0 - 任务的已尝试次数大于 命令行的
--tries参数 - 开发者添加了
queue_failed事件标签及其对应的回调代码 - 消费者类中定义了
failed()方法，用于接收任务失败的通知