Yii2 队列 shmilyzxt/yii2-queue 简单概述
shmilyzxt/yii2-queue简单解释:
1.我用的yii2高级版,我们从配置开始看代码,这里我用的是mysql队列,首先配置文件,我把queue配置项写在根目录common\config\main-local.php下的components数组下,更改一下数据库配置.复制composer安装后复制
vendor\shmilyzxt\yii2-queue\jobs\jobs.sql vendor\shmilyzxt\yii2-queue\failed\failed.sql
2个sql文件到数据库中建立队列数据表和执行任务失败时的数据表.
2.推送任务开始语法:\Yii::$app->queue->pushOn(newSendMial(),['email'=>'49783121@qq.com','title'=>'test','content'=>'emailtest'],'email');我们到vendor\shmilyzxt\queue\queues\DatabaseQueue.php去看看代码,pushOn()方法写在了DatabaseQueue类的父类vendor\shmilyzxt\queue\base\Queue.php中:
//入队列
publicfunctionpushOn($job,$data='',$queue=null)
{
//canPush检查队列是否已达最大任务量
if($this->canPush()){
//beforePush入队列前的事件
$this->trigger(self::EVENT_BEFORE_PUSH);
//入队列
$ret=$this->push($job,$data,$queue);
//afterPush入队列后的事件
$this->trigger(self::EVENT_AFTER_PUSH);
return$ret;
}else{
thrownew\Exception("maxjobsnumberexceed!themaxjobsnumberis{$this->maxJob}");
}
}
注释:这里最好去看看yii2event事件类,http://www.digpage.com/event.html
关于入队列:$this->push($job,$data,$queue);,这里在配合queue类文件查看,相关函数跳转,处理一下数据记录到数据库中.(函数走向:getQueue()-->createPayload()-->pushToDatabase()),pushOn()最终返回数据插入数据库的结果,成功$ret是1.
3.后台运行命令处理队列,例:php./yiiworker/listendefault1012830其中default是队列的名称,上面推送了一个email队列应该改为email.
启动命令后,我们来看代码:首先执行:WorkerController控制器actionListen方法,我们跟着代码进入到vendor\shmilyzxt\queue\Worker.php--listen方法中,这里其实就是一直在循环,执行操作队列的任务:
/**
*启用一个队列后台监听任务
*@paramQueue$queue
*@paramstring$queueName监听队列的名称(在pushon的时候把任务推送到哪个队列，则需要监听相应的队列才能获取任务)
*@paramint$attempt队列任务失败尝试次数，0为不限制
*@paramint$memory允许使用的最大内存
*@paramint$sleep每次检测的时间间隔
*/
publicstaticfunctionlisten(Queue$queue,$queueName='default',$attempt=10,$memory=512,$sleep=3,$delay=0){
while(true){
try{
//DatabaseQueue从数据库队列取出一个可用任务(实例),并且更新任务
$job=$queue->pop($queueName);
}catch(\Exception$e){
throw$e;
continue;
}
if($jobinstanceofJob){
//判断执行错误的次数是否大于传入的执行次数
if($attempt>0&&$job->getAttempts()>$attempt){
$job->failed();
}else{
try{
//thrownew\Exception("testfailed");
$job->execute();
}catch(\Exception$e){
//执行失败,判断是否被删除,重新入队
if(!$job->isDeleted()){
$job->release($delay);
}
}
}
}else{
self::sleep($sleep);
}
if(self::memoryExceeded($memory)){
self::stop();
}
}
}
注释:在$queue->pop($queueName);是vendor\shmilyzxt\queue\queues\DatabaseQueue.php方法内使用事务执行SQL,并且创建vendor\shmilyzxt\queue\jobs\DatabaseJob.php的实例
//取出一个任务
publicfunctionpop($queue=null)
{
$queue=$this->getQueue($queue);
if(!is_null($this->expire)){
//$this->releaseJobsThatHaveBeenReservedTooLong($queue);
}
$tran=$this->connector->beginTransaction();
//判断是否有一个可用的任务需要执行
if($job=$this->getNextAvailableJob($queue)){
$this->markJobAsReserved($job->id);
$tran->commit();
$config=array_merge($this->jobEvent,[
'class'=>'shmilyzxt\queue\jobs\DatabaseJob',
'queue'=>$queue,
'job'=>$job,
'queueInstance'=>$this,
]);
return\Yii::createObject($config);
}
$tran->commit();
returnfalse;
}
至于:$job->execute();是DatabaseJob继承父类Job执行的,顺着代码找下去是yii\base\Componenttrigger执行的事件,
/**
*执行任务
*/
publicfunctionexecute()
{
$this->trigger(self::EVENT_BEFORE_EXECUTE,newJobEvent(["job"=>$this,'payload'=>$this->getPayload()]));//beforeExecute执行任务之前的一个事件在JobEvent中并没有什么可执行的代码
$this->resolveAndFire();//真正执行的任务的方法
}
/**
*真正任务执行方法（调用hander的handle方法）
*@paramarray$payload
*@returnvoid
*/
protectedfunctionresolveAndFire()
{
$payload=$this->getPayload();
$payload=unserialize($payload);//反序列化数据
$type=$payload['type'];
$class=$payload['job'];
if($type=='closure'&&($closure=(newSerializer())->unserialize($class[1]))instanceof\Closure){
$this->handler=$this->getHander($class[0]);
$this->handler->closure=$closure;
$this->handler->handle($this,$payload['data']);
}elseif($type=='classMethod'){
$payload['job'][0]->$payload['job'][1]($this,$payload['data']);
}elseif($type=='staticMethod'){
$payload['job'][0]::$payload['job'][1]($this,$payload['data']);
}else{//执行的`SendMail`类的`handle($job,$data)`方法
$this->handler=$this->getHander($class);
$this->handler->handle($this,$payload['data']);
}
//执行完任务后删除
if(!$this->isDeletedOrReleased()){
$this->delete();
}
}
最后到了执行的SendMail类的handle($job,$data),在这里就是推送到队列的对象和数据,接着就是我们的处理逻辑了.
publicfunctionhandle($job,$data)
{
if($job->getAttempts()>3){
$this->failed($job);
}
$payload=$job->getPayload();
echo'';print_r($payload);
//$payload即任务的数据，你拿到任务数据后就可以执行发邮件了
//TODO发邮件
}
总结
以上所述是小编给大家介绍的Yii2队列shmilyzxt/yii2-queue简介，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对毛票票网站的支持！