PHPyii2 대기열 소개 shmilyzxt/yii2-queue

• 사용하시면서 꽤 좋다고 생각합니다.

• shmilyzxt/yii2-queue 간단한 설명: shmilyzxt/yii2-queue 简单解释:

1. 我用的yii2高级版,我们从配置开始看代码,这里我用的是mysql队列,首先配置文件,我把queue配置项写在根目录commonconfigmain-local.php下的 components数组下,更改一下数据库配置.复制composer安装后复制
   vendorshmilyzxtyii2-queuejobsjobs.sql
vendorshmilyzxtyii2-queuefailedfailed.sql
   2个sql文件到数据库中建立队列数据表和执行任务失败时的数据表.

2. 推送任务开始语法:Yii::$app->queue->pushOn(new SendMial(),['email'=>'49783121@qq.com','title'=>'test','content'=>'email test'],'email'); 我们到vendorshmilyzxtqueuequeuesDatabaseQueue.php去看看代码,pushOn()方法写在了DatabaseQueue类的父类vendorshmilyzxtqueuebaseQueue.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事件类.

关于入队列: $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方法,我们跟着代码进入到 vendorshmilyzxtqueueWorker.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);是vendorshmilyzxtqueuequeuesDatabaseQueue.php方法内使用事务执行SQL,并且创建vendorshmilyzxtqueuejobsDatabaseJob.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执行的,顺着代码找下去是yiibaseComponent trigger执行的事件,

/** 
 * 执行任务 
 */
public function execute(){   
  //beforeExecute 执行任务之前的一个事件 在JobEvent中并没有什么可执行的代码
  $this->trigger(self::EVENT_BEFORE_EXECUTE, new JobEvent(["job" => $this, 'payload' => $this->getPayload()]));
  $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)

저는 yii2 고급 버전을 사용합니다. 먼저 구성부터 살펴보겠습니다. 여기서는 mysql 대기열을 사용합니다. 먼저 대기열을 작성합니다. code> 구성 항목을 루트 디렉토리 <code>commonconfigmain-local.php의 comminors 배열 아래에서 설치 후 composer를 복사하세요. vendorshmilyzxtyii2- queuejobsjobs.sql

vendorshmilyzxtyii2-queuefailedfailed.sql🎜2개의 sql 파일을 데이터베이스에 추가하여 작업 실패 시 대기열 데이터 테이블과 데이터 테이블을 설정합니다.🎜🎜🎜 🎜푸시 작업 시작 구문: Yii::$app->queue->pushOn(new SendMial(),['email'=>'49783121@qq.com','title'=>' test','content' =>'email test'],'email'); vendorshmilyzxtqueuequeuesDatabaseQueue.php로 이동하여 pushOn() 메소드는 DatabaseQueue 클래스 vendorshmilyzxtqueuebaseQueue.php에 작성됩니다. 🎜🎜

🎜🎜🎜

public function handle($job,$data)
    {        if($job->getAttempts() > 3){            $this->failed($job);
        }        $payload = $job->getPayload();
        echo &#39;<pre class="brush:php;toolbar:false">&#39;;print_r($payload);
        //$payload即任务的数据，你拿到任务数据后就可以执行发邮件了
        //TODO 发邮件
    }

🎜참고: 사용하는 것이 가장 좋습니다. 여기서 yii2 이벤트 이벤트 클래스를 살펴보세요.🎜🎜큐 입력 정보: $this->push($job, $data, $queue);, 여기서는 queue 클래스 파일을 열고 관련 함수로 점프하여 처리합니다. (함수 방향: getQueue()-->createPayload()-->pushToDatabase( )), pushOn()은 최종적으로 데이터 삽입을 반환합니다. 데이터베이스의 성공적인 $ret 결과는 1입니다.🎜🎜3. 백그라운드 실행 명령 처리 대기열. 예: php ./yii 작업자/listen default 10 128 3 0 여기서 default는 위에 푸시된 email 대기열의 이름입니다. 명령을 시작한 후 코드를 살펴보겠습니다. 먼저 WorkerController 컨트롤러 actionListen 메서드를 실행하고 코드를 따릅니다. 그리고 vendorshmilyzxtqueueWorker.php -- listening 메소드를 입력하세요. 이는 실제로 작업을 계속 실행하는 루프입니다. 🎜🎜🎜🎜rrreee🎜 참고: $queue->pop( $queueName);는 SQL을 실행하기 위해 트랜잭션을 사용하는 vendorshmilyzxtqueuequeuesDatabaseQueue.php 메소드이며, vendorshmilyzxtqueuejobsDatabaseJob.php🎜🎜🎜🎜rrreee🎜의 인스턴스를 생성합니다. : $job->execute();는 DatabaseJob이 상위 클래스Job를 상속받아 실행된다는 것입니다. 코드를 따라 실행된 이벤트를 찾으세요. yiibaseComponent Trigger에 의해, 🎜🎜🎜🎜rrreee🎜🎜🎜rrreee🎜마지막으로 실행된 SendMailhandle($job,$data)에 도달합니다. > 클래스의 개체와 데이터가 대기열에 푸시된 다음 처리 논리가 있습니다.🎜🎜🎜🎜rrreee

위 내용은 PHPyii2 대기열 소개 shmilyzxt/yii2-queue의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!