Development

Laravel Parallel Job Processing Tip.

By Pete Wond

2 minutes

Whilst recently working on porting a legacy client application to Laravel we needed to implement a pipeline of jobs to generate a temporary file, run several processes on the file and then remove the file afterwards. As per the Laravel docs, this is possible via Job Chaining which:

“Job chaining allows you to specify a list of queued jobs that should be run in sequence after the primary job has executed successfully”

In this use case we a Job class ProcessFileJob that dispatches a chain of Jobs that will be worked through in order:

php

Copied

 1final class ProcessFileJob implements ShouldQueue
 2{
 3    ...
 4    public function handle(): void
 5    {
 6        Bus::chain([
 7            new CreateTemporaryFileJob(),
 8            new DoSomethingToFileJob(),
 9            new DoSomethingElseToFileJob(),
10            new CleanUpAfterWardsJob(),
11        ])->dispatch();
12    }
13}

final class ProcessFileJob implements ShouldQueue
{
    ...
    public function handle(): void
    {
        Bus::chain([
            new CreateTemporaryFileJob(),
            new DoSomethingToFileJob(),
            new DoSomethingElseToFileJob(),
            new CleanUpAfterWardsJob(),
        ])->dispatch();
    }
}

This is fine and works but it became apparent that DoSomethingToFileJob and DoSomethingElseToFileJob are not dependent on each other. Further as they are both long running tasks it would be nice if they could run in parallel so we can potentially clean up sooner! We can achieve this by delegating the processing of the parallel job chains into another Job with its own batch like this:

php

Copied

 1final class ProcessFileJob implements ShouldQueue
 2{
 3    ...
 4    public function handle(): void
 5    {
 6        Bus::chain([
 7            new CreateTemporaryFileJob(),
 8            new HandleParallelJobs(), // replace two blocking jobs with one parallel handler
 9            new CleanUpAfterWardsJob(),
10        ])->dispatch();
11    }
12}
13final class HandleParallelJobs implements ShouldQueue
14{
15    use Dispatchable, InteractsWithQueue, Queueable;
16    public function handle(): void
17    {
18        $parallelJobChains = [
19            [
20                new DoSomethingToFileJob(),
21            ],
22            [
23                new DoSomethingElseToFileJob(),
24            ],
25        ];
26        Bus::batch($parallelJobChains)->dispatch();
27    }
28}

final class ProcessFileJob implements ShouldQueue
{
    ...
    public function handle(): void
    {
        Bus::chain([
            new CreateTemporaryFileJob(),
            new HandleParallelJobs(), // replace two blocking jobs with one parallel handler
            new CleanUpAfterWardsJob(),
        ])->dispatch();
    }
}
final class HandleParallelJobs implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable;
    public function handle(): void
    {
        $parallelJobChains = [
            [
                new DoSomethingToFileJob(),
            ],
            [
                new DoSomethingElseToFileJob(),
            ],
        ];
        Bus::batch($parallelJobChains)->dispatch();
    }
}

Moving the parallel process into another Job means we can verify this is behaving as expected through a discrete and succinct unit test like this:

php

Copied

 1test('job creates batch with parallel chains as expected', function () {
 2    Bus::fake([
 3        DoSomethingToFileJob::class,
 4        DoSomethingElseToFileJob::class,
 5    ]);
 6    HandleParallelJobs::dispatch();
 7    Bus::assertBatched(function (PendingBatch $batch) {
 8        // assert that we have two chains
 9        return $batch->jobs->count() === 2
10            // assert one chain for DoSomethingToFileJob
11            && is_array($batch->jobs[0])
12            && $batch->jobs[0][0] instanceof DoSomethingToFileJob
13            // assert one chain for DoSomethingElseToFileJob
14            && is_array($batch->jobs[1])
15            && $batch->jobs[1][0] instanceof DoSomethingElseToFileJob;
16    });
17});

test('job creates batch with parallel chains as expected', function () {
    Bus::fake([
        DoSomethingToFileJob::class,
        DoSomethingElseToFileJob::class,
    ]);
    HandleParallelJobs::dispatch();
    Bus::assertBatched(function (PendingBatch $batch) {
        // assert that we have two chains
        return $batch->jobs->count() === 2
            // assert one chain for DoSomethingToFileJob
            && is_array($batch->jobs[0])
            && $batch->jobs[0][0] instanceof DoSomethingToFileJob
            // assert one chain for DoSomethingElseToFileJob
            && is_array($batch->jobs[1])
            && $batch->jobs[1][0] instanceof DoSomethingElseToFileJob;
    });
});

So now our Jobs can be process more efficiently without un-necessary blocking, nice! What patterns do you use to process complicated Job flows? What ways would you test they’re working as you’d expect? Let us know on Twitter or Linkedin!

Laravel Parallel Job Processing Tip.

Laravel Partner

Vue Experts