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 队列 shmilyzxt/yii2-queue简介,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对我们网站的支持!

(0)

相关推荐

  • 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个sq

  • Oracle高级队列(Advanced Queue)简单实例

    Oracle高级队列(Advanced Queue)简单实例 最简单的高级队列的练习,基本上都使用了默认参数. -- Create Type create or replace type note as Object( subject varchar2(100), Content varchar2(2000),CreateTime date ); -- Create queue table begin sys.dbms_aqadm.create_queue_table( queue_table

  • yii2.0框架场景的简单使用示例

    本文实例讲述了yii2.0框架场景的简单使用.分享给大家供大家参考,具体如下: 一.规则中使用场景 规则场景的使用 模型层 public function rules() { return [ [['name','product_id'],'required','on'=>'add], ]; } 表示只在add场景下改规则才生效. 在控制器中指定场景: $model->scenario = "add"; 或者 $model = new User(['scenario' =&

  • YII2框架中验证码的简单使用方法示例

    本文实例讲述了YII2框架中验证码的简单使用方法.分享给大家供大家参考,具体如下: 验证码的使用是比较频繁的.YII2中已经帮我们做好了封装. 首先我们在控制器里创建一个actions方法,用于使用yii\captcha\CaptchaAction <?php namespace app\controllers; use YII; use yii\web\Controller; class IndexController extends Controller { public function

  • Laravel 4.2 中队列服务(queue)使用感受

    这半个月,我参与重写了一个微信公众号后端系统,首次使用了laravel 4.2,以及laravel引以为傲的队列服务(queue). 由于整个系统涉及到多端交互,又有大量语音传输.处理的业务,我们在一些地方发现响应时间过长.之前的系统基于node.js和mongoDB,由于node天生就是异步,有守护进程,所以并没有出现过这个问题,而这次重写必然要引入异步流程了.Queue进入了我们的视线. 根据这一页几乎还全是英文的"中文文档" ,laravel恰好在4.2版本中刚刚引入了redis

  • Java 阻塞队列详解及简单使用

     Java 阻塞队列详解 概要: 在新增的Concurrent包中,BlockingQueue很好的解决了多线程中,如何高效安全"传输"数据的问题.通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利.本文详细介绍了BlockingQueue家庭中的所有成员,包括他们各自的功能以及常见使用场景. 认识BlockingQueue阻塞队列,顾名思义,首先它是一个队列,而一个队列在数据结构中所起的作用大致如下图所示: 从上图我们可以很清楚看到,通过一个共享的队列,

  • Yii2.0建立公共方法简单示例

    本文实例讲述了Yii2.0建立公共方法.分享给大家供大家参考,具体如下: 因为之前写项目都是用的Thinkphp,公共方法都写在Tp自带的common下面的function里面,初次接触Yii框架发现它自身没带这个机制.在项目中,很多地方都需要公共方法的存在,以提高代码复用性,减少重复开发的时间,那yii2如何定义自己的公共方法呢? 以yii2-advanced版本为基础来进行介绍: 上图为yii2-advanced版本的基础目录结构,而./common 目录为公共资源目录,我们可以把定义的方法

  • Python 进程操作之进程间通过队列共享数据,队列Queue简单示例

    本文实例讲述了Python 进程操作之进程间通过队列共享数据,队列Queue.分享给大家供大家参考,具体如下: 队列中的数据是放在内存中的,可以通过分布式缓存redis优化队列. demo.py(进程通过队列共享数据): import multiprocessing def download_from_web(q): """下载数据""" # 模拟从网上下载的数据 data = [11, 22, 33, 44] # 向队列中写入数据 for te

  • yii2.0框架数据库操作简单示例【添加,修改,删除,查询,打印等】

    本文实例讲述了yii2.0框架数据库操作.分享给大家供大家参考,具体如下: 添加 $id = \Yii::$app->db ->createCommand() ->insert('表名',['car_num' => $car_num, 'lg_shop_id' => $shop_id]) ->execute(); batchInsert():一次添加多行 // table name, column names, column values Yii::$app->d

  • Java 队列实现原理及简单实现代码

    Java 队列实现原理 "队列"这个单词是英国人说的"排".在英国"排队"的意思就是站到一排当中去.计算机科学中,队列是一种数据结构,有点类似栈,只是在队列中第一个插入的数据项也会最先被移除,而在栈中,最后插入的数据项最先移除.队列的作用就像电影院前的人们站成的排一样:第一个进入附属的人将最先到达队头买票.最后排队的人最后才能买到票. 队列和栈一样也被用作程序员的工具.它也可以用于模拟真实世界的环境,例如模拟人们在银行里排队等待,飞机等待起飞,或

随机推荐