Laravel 嵌套作业

Posted 2023-02-22

【中文标题】Laravel 嵌套作业

【英文标题】：Laravel Nested Jobs

【发布时间】：2019-11-13 14:25:30

【问题描述】：

我创建了一个具有分派另一个作业的 foreach 循环的作业。有没有办法在所有嵌套作业都完成后触发？

这里触发时会发生什么

第 1 步。首先我触发批处理作业 GenerateBatchReports::dispatch($orderable);

第 2 步。然后我们运行一个循环并将其他作业排队

/**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    
        $dir = storage_path('reports/tmp/'.str_slug($this->event->company) . '-event');

        if(file_exists($dir)) 
            File::deleteDirectory($dir);
        

        foreach($this->event->participants as $participant) 
            $model = $participant->exercise;

            GenerateSingleReport::dispatch($model);
        
    

我只需要知道所有嵌套作业何时完成，这样我就可以压缩报告并将它们通过电子邮件发送给用户。当批处理作业完成对所有嵌套作业的排队时，它会从列表中删除。有没有办法在嵌套作业完成之前保留作业，然后触发事件？

任何帮助将不胜感激。

【问题讨论】：

您可以同步运行子作业，这将使父作业保持活动状态，直到它们全部完成：laravel.com/docs/5.8/queues#synchronous-dispatching

谢谢。这看起来很有希望，但我的应用程序仍在使用 laravel 5.5。看起来 DispatchNow 方法直到 5.7 才引入。

在这种情况下，您可以使用sync。检查我答案的下半部分。

在 Laravel 队列文档中搜索作业链。它解决了您的问题，但您的作业不会并行运行（如果使用同步也会发生同样的情况）

在上一个作业结束时触发自定义事件有何不同？

【参考方案1】：

对于 laravel >= 5.7

您可以使用dispatchNow 方法。这将使父作业在处理子作业时保持活动状态：

https://laravel.com/docs/5.8/queues#synchronous-dispatching

父母工作：

public function handle()

    // ...

    foreach($this->event->participants as $participant) 
        $model = $participant->exercise;

        GenerateSingleReport::dispatchNow($model);
    

    // then do something else...

适用于 laravel 5.2 - 5.6

您可以使用sync 连接：

https://laravel.com/docs/5.5/queues#customizing-the-queue-and-connection

确保在您的config/queue.php 中定义了连接：

https://github.com/laravel/laravel/blob/5.5/config/queue.php#L31

父作业（注意：此语法适用于 5.5。5.2 的文档略有不同）：

public function handle()

    // ...

    foreach($this->event->participants as $participant) 
        $model = $participant->exercise;

        GenerateSingleReport::dispatch($model)->onConnection('sync');
    

    // then do something else...

【讨论】：

使用sync 与使用队列在后台处理作业的想法相矛盾。永远不要在生产环境中这样做。

@Namoshek 父作业仍在异步队列中，因此同步的子作业仍将在主请求周期之外运行。

这很有趣，我需要试试这个。应该可以构建与我猜想的事件非常相似的东西（一个调度的后台作业 + 带有非排队处理程序的事件，其中每个处理程序通过一个事件触发下一个处理程序）。

【参考方案2】：

你可以使用 Laravel 的 job chaining。它允许您按顺序运行一堆作业，如果一个失败，则链中的其余作业将不会运行。

基本语法如下所示：

FirstJob::withChain([
    new SecondJob($param),
    new ThirdJob($param)
])->dispatch($param_for_first_job);

在您的情况下，您可以将所有 GenerateSingleReport 作业添加到数组中，除了第一个作业，然后添加然后将您要运行的最终作业添加到数组的末尾。然后您可以将该数组传递给第一个作业的withChain 方法。

$jobs = [];
$first_job = null;
$first_parameter = null;

foreach($this->event->participants as $participant) 
    $model = $participant->exercise;

    if (empty($first_job)) 
        $first_job = GenerateSingleReport;
        $first_parameter = $model;
     else 
        $jobs[] = new GenerateSingleReport($model);
                


$jobs[] = new FinalJob();

$first_job->withChain($jobs)->dispatch($first_parameter);

【参考方案3】：

更新：Laravel 8（计划于 2020 年 9 月 8 日发布）将提供作业批处理。这个功能是already documented 可能非常适合嵌套作业场景，如下所示：

$batch = Bus::batch([
    new ProcessPodcast(Podcast::find(1)),
    new ProcessPodcast(Podcast::find(2)),
    new ProcessPodcast(Podcast::find(3)),
    new ProcessPodcast(Podcast::find(4)),
    new ProcessPodcast(Podcast::find(5)),
])->then(function (Batch $batch) 
    // All jobs completed successfully...
)->catch(function (Batch $batch, Throwable $e) 
    // First batch job failure detected...
)->finally(function (Batch $batch) 
    // The batch has finished executing...
)->dispatch();

我们还将能够即时添加其他批处理作业：

$this->batch()->add(Collection::times(1000, function () 
    return new ImportContacts;
));

原答案?

我想出了一个不同的解决方案，因为我有一个使用多个进程的队列。所以，对我来说：

没有dispatchNow，因为我想保持作业并行运行。 有多个进程，我需要确保最后一个嵌套作业不会在最后一个嵌套作业之后运行。因此，简单的链接并不能保证这一点。

所以我满足要求的不优雅的解决方案是分派所有嵌套作业，并在最后一个中分派最终作业，延迟几秒钟，以确保所有其他可能仍在运行的嵌套作业并行将被终止。

/**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    
        $last_participant_id = $this->event->participants->last()->id;

        foreach($this->event->participants as $participant) 
            $is_last = $participant->id === $last_participant_id;

            GenerateSingleReport::dispatch($model, $is_last);
        
    

在GenerateSingleReport.php

class GenerateSingleReport implements ShouldQueue

    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $model;
    protected $runFinalJob;

    public function __construct($model, $run_final_job = false)
    
        $this->model = $model;
        $this->runFinalJob = $run_final_job;
    

    public function handle()
    
        // job normal stuff…

        if ($this->runFinalJob) 
            FinalJob::dispatch()->delay(30);
        
    

或者

我提出了另一个想法，因此代码并非完美无缺。也许可以创建一个包装器 Job 并专门用于运行与最终作业链接的最后一个嵌套作业。

/**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    
        $last_participant_id = $this->event->participants->last()->id;

        foreach($this->event->participants as $participant) 
            $is_last = $participant->id === $last_participant_id;

            if ($is_last) 
                ChainWithDelay::dispatch(
                    new GenerateSingleReport($model), // last nested job
                    new FinalJob(), // final job
                    30 // delay
                );
             else 
                GenerateSingleReport::dispatch($model, $is_last);
            
        
    

在ChainWithDelay.php

class ChainWithDelay implements ShouldQueue

    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $job;
    protected $finalJob;
    protected $delay;

    public function __construct($job, $final_job, $delay = 0)
    
        $this->job = $job;
        $this->finalJob = $final_job;
        $this->delay = $delay;
    

    public function handle()
    
        $this->job
            ->withChain($this->finalJob->delay($this->delay))
            ->dispatchNow();