本文主要介绍了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(new sendmial(),['email'=>'49783121@qq.com','title'=>'test','content'=>'email test'],'email'); 我们到vendor\shmilyzxt\queue\queues\databasequeue.php去看看代码,pushon()方法写在了databasequeue类的父类vendor\shmilyzxt\queue\base\queue.php中:
//入队列
public function pushon($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 {
throw new \exception("max jobs number exceed! the max jobs number is {$this->maxjob}");
}
}
注释:这里最好去看看yii2 event事件类,http://www.digpage.com/event.html
关于入队列: $this->push($job, $data, $queue);,这里在配合queue类文件查看,相关函数跳转,处理一下数据记录到数据库中.(函数走向:getqueue()-->createpayload()-->pushtodatabase()),pushon()最终返回数据插入数据库的结果,成功$ret是1.
3.后台运行命令处理队列,例:php ./yii worker/listen default 10 128 3 0 其中default是队列的名称,上面推送了一个email队列 应该改为email.
启动命令后,我们来看代码:首先执行:workercontroller控制器 actionlisten方法,我们跟着代码进入到 vendor\shmilyzxt\queue\worker.php -- listen方法中,这里其实就是一直在循环,执行操作队列的任务:
/**
* 启用一个队列后台监听任务
* @param queue $queue
* @param string $queuename 监听队列的名称(在pushon的时候把任务推送到哪个队列，则需要监听相应的队列才能获取任务)
* @param int $attempt 队列任务失败尝试次数，0为不限制
* @param int $memory 允许使用的最大内存
* @param int $sleep 每次检测的时间间隔
*/
public static function listen(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($job instanceof job){
//判断执行错误的次数是否大于传入的执行次数
if($attempt > 0 && $job->getattempts() > $attempt){
$job->failed();
}else{
try{
//throw new \exception("test failed");
$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的实例
//取出一个任务
public function pop($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();
return false;
}
至于:$job->execute();是databasejob继承父类job执行的,顺着代码找下去是yii\base\component trigger执行的事件,
/**
* 执行任务
*/
public function execute()
{
$this->trigger(self::event_before_execute, new jobevent(["job" => $this, 'payload' => $this->getpayload()]));//beforeexecute 执行任务之前的一个事件 在jobevent中并没有什么可执行的代码
$this->resolveandfire();//真正执行的任务的方法
}
/**
* 真正任务执行方法（调用hander的handle方法）
* @param array $payload
* @return void
*/
protected function resolveandfire()
{
$payload = $this->getpayload();
$payload = unserialize($payload); //反序列化数据
$type = $payload['type'];
$class = $payload['job'];
if ($type == 'closure' && ($closure = (new serializer())->unserialize($class[1])) instanceof \closure) {
$this->handler = $this->gethander($class[0]);
$this->handler->closure = $closure;
$this->handler->handle($this, $payload['data']);
} else if ($type == 'classmethod') {
$payload['job'][0]->$payload['job'][1]($this, $payload['data']);
} else if ($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),在这里就是推送到队列的对象和数据,接着就是我们的处理逻辑了.
public function handle($job,$data)
{
if($job->getattempts() > 3){
$this->failed($job);
}
$payload = $job->getpayload();
echo '<pre>';print_r($payload);
//$payload即任务的数据，你拿到任务数据后就可以执行发邮件了
//todo 发邮件
}
相关推荐：
yii2整合迅搜实现高效中文分词检索
yii解决deleteall连表删除报错问题
yii如何过滤不良代码
以上就是简述yii2队列shmilyzxt/yii2-queue的详细内容。