Laravelのキューイング

私の現在のプロジェクトには、バックグラウンドで実行される多くのタスクがあります。 データは外部サービスから到着し、処理のいくつかの段階を経ます。 処理はキューメカニズムを介して実装されます。 これは便利です。プロセスのタイプごとにワーカーの数を変えることができます。 そして、何かが落ちた場合、キューは蓄積され、データは失われません-問題が修正されるとすぐに処理されます。



処理の次の段階のために1つのプロセスからタスクを作成するには、処理の最後に単純にdispatch()



と呼びます。次のようなものです。

 class MyFirstJob extends Job { use DispatchesJobs; protected $data; public function __construct($data) { $this->data = $data; } public function handle() { $this->doSomething($this->data); $this->dispatch(new MySecondJob($this->data)); // Second task } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、処理の次の段階はまったく同じ方法で開始されました。

 class MySecondJob extends Job { use DispatchesJobs; protected $data; public function __construct($data) { $this->data = $data; } public function handle() { $this->doSomething($this->data); if ($this->someCondition($this->data)) { $this->dispatch(new MyThirdJob($this->data)); // Third task } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

最初は良かったのですが、新しい処理ステップが追加され、チェーンが成長しました。 もう一度、別の処理段階（新しい行）を追加する必要が生じたとき、処理中の処理内容と順序を正確に思い出せなくなると思いました。 そして、これを理解するためのコードはそれほど単純ではありません。 ビジネスロジックの要素が登場しました。そのような場合、そのような処理が開始され、別の場合には、一連のタスクがすぐに作成されます。 一般的に、大規模システムで見るのが「大好き」なものすべて。



ああ、何かをする時だと思った。 そして、処理順序（ dispatch()



呼び出しの順序）を個別のコードに制御することが非常に便利であると判断しました。 その後、すべてが論理的で明確になります-ここにはビジネスプロセス（制御コード、キューマネージャー）があり、ここには個別の部分（キュー）があります。



私はそれをしましたが、まだ満足しています。 今、私がしたことをお話しします。 このアプローチがあなたにも役立つなら、私はうれしいです。

キュー管理

複数の独立したデータ処理プロセスがあります。 各アルゴリズムを個別に説明するために、キューマネージャーの抽象クラスを作成します。

 <?php namespace App\Jobs\Pipeline; use App\Jobs\Job; use Illuminate\Foundation\Bus\DispatchesJobs; abstract class PipelineAbstract { use DispatchesJobs; /** * @param array $params * @return PipelineAbstract */ public function start(array $params) { $this->next(null, $params); return $this; } /** * @param Job $currentJob * @param array $params Set of parameters for starting new jobs */ abstract public function next(Job $currentJob = null, array $params); /** * @param Job $job */ protected function startJob(Job $job) { $this->dispatch($job); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

next()



メソッドでは、ビジネスプロセスを実装します。 startJob()



は、 startJob()



dispatch()



単なるラッパーです。 そして、データ処理のプロセス全体を開始する必要がある場所（外部サービスからのデータが到着する場所start()



でstart()



を使用します。



ビジネスロジックの実装例：

 <?php namespace App\Jobs\Pipeline; use App\Jobs\Job; use App\Jobs\MyFirstJob; use App\Jobs\MySecondJob; use App\Jobs\MyThirdJob; class ProcessDataPipeline extends PipelineAbstract { /** * @inheritdoc */ public function next(Job $currentJob = null, array $params) { // Start first job if ($currentJob === null) { $this->startJob(new MyFirstJob($params, $this)); } if ($currentJob instanceof MyFirstJob) { $this->startJob(new MySecondJob($params, $this)); } if ($currentJob instanceof MySecondJob) { if ($this->someCondition($params)) { $this->startJob(new MyThirdJob($params, $this)); } } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

以上です。 MyFirstJob



の起動に置き換わるだけです。



だった

 $this->dispatch(new MyFirstJob($data));
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

になっています

 (new ProcessDataPipeline())->start($data);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、残りのキューにタスクを追加する代わりに、 next()



メソッドを呼び出します。



だった

 $this->dispatch(new MySecondJob($data));
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

になっています

 $this->next($data);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

私はほとんど忘れていました。 このために、キューの基本クラスを変更する必要があります。 上記のコードでは、キューオブジェクトをインスタンス化するときに、データに加えてパイプラインオブジェクトも転送することは明らかです。

 <?php namespace App\Jobs; use App\Jobs\Pipeline\PipelineAbstract; abstract class Job { /** * @param array $params */ public function next(array $params) { if ($this->pipeline) { $this->pipeline->next($this, $params); } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

また、特定のジョブのコンストラクターでは、パイプラインインスタンスを受け入れて、ビジネスロジックステップ（ next()



メソッドの呼び出し）が目的のパイプライン実装によって処理されるようにします。

 class MyFirstJob extends Job { /** * @param mixed data * @param PipelineAbstract|null $pipeline */ public function __construct($data, PipelineAbstract $pipeline = null) { $this->data = $data; $this->pipeline = $pipeline; } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これですべてです。 責任の連鎖のようになりました。 アイデアを簡単な言葉で説明しようとしました。 あなたが突然これをやりたいと思った場合、私は実装の実用的な例を公開しました 、多分それは言葉よりも誰かにとってより便利でしょう：

なんて良い

• データ処理プロセスの説明は、コードではなく、1つの方法に集中しています。
• キュー管理メカニズムに新しい動作をきちんと追加できるようになりました。 たとえば、ロギング、各ステップの処理状態のデータベースへの保存。
• 新しい処理ステップの追加とタスクの順序の変更が簡単になりました。

ところで、Laravelの最新バージョンでは、 withChain()



た同様のツールwithChain()



、タスクが厳密な順序で実行されることを保証します。 簡単な場合、これで十分です。 ただし、特定のプロセスを起動する条件がある場合、次のプロセスのデータが前のプロセスで生成されると、より普遍的なメカニズムが依然として必要になります。 たとえば、この記事で私が話したもの。