Batches, Chains and Unique Jobs

Grouping jobs into a batch with callbacks, running them in order with a chain, deduplicating with ShouldBeUnique, and job middleware.

SyntaxBus::batch(array $jobs)->then(...)->catch(...)->finally(...)->dispatch(): Illuminate\Bus\Batch

Three shapes cover most work that is more than one job.

Batches

A batch is a group of jobs tracked as one thing, with callbacks that fire when the group finishes. Jobs in it must use the Illuminate\Bus\Batchable trait. Bus::batch([...]) takes then() for all-succeeded, catch() for the first failure and finally() for either, then dispatch() returns an Illuminate\Bus\Batch whose id you can store and look up later with Bus::findBatch(). Progress lives in the job_batches table, created by php artisan make:queue-batches-table. By default one failure cancels the rest; allowFailures() lets the others continue.

Chains

Bus::chain([...]) runs jobs one after another, each dispatched only when the previous one succeeded. If any job in the chain fails, the remainder never run, and catch() on the chain is called. Use a chain for steps, a batch for work that can run in parallel.

Unique jobs

A job implementing Illuminate\Contracts\Queue\ShouldBeUnique takes a cache lock keyed on uniqueId() when dispatched; a second dispatch with the same id is dropped silently. #[UniqueFor(3600)] sets how long the lock survives, so a crashed worker cannot block the job for ever. ShouldBeUniqueUntilProcessing releases the lock when the job starts instead of when it finishes.

Job middleware

php artisan make:job-middleware RateLimited generates a class with handle(object $job, Closure $next), and a middleware() method on the job returns the instances to wrap it in. WithoutOverlapping and RateLimited ship with the framework.

Example

Example · php
<?php
// app/Jobs/ImportCsvChunk.php

declare(strict_types=1);

namespace App\Jobs;

use App\Models\Import;
use Illuminate\Bus\Batchable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\Tries;
use Illuminate\Queue\Middleware\WithoutOverlapping;

#[Tries(5)]
class ImportCsvChunk implements ShouldQueue
{
    use Batchable, Queueable;

    public function __construct(
        public int $importId,
        public int $firstRow,
        public int $lastRow,
    ) {}

    /**
     * Only one chunk of a given import writes at a time. A chunk that
     * cannot take the lock is put back after 30 seconds, which costs
     * it one of the five attempts above.
     */
    public function middleware(): array
    {
        return [
            (new WithoutOverlapping("import:{$this->importId}"))
                ->releaseAfter(30)
                ->expireAfter(180),
        ];
    }

    public function handle(): void
    {
        // Another chunk failed and cancelled the batch: stop.
        if ($this->batch()?->cancelled()) {
            return;
        }

        Import::findOrFail($this->importId)
            ->importRows($this->firstRow, $this->lastRow);
    }
}

When to use it

  • Import a large CSV as 500-row chunks and mark the import complete when all of them finish.
  • Charge, then render the PDF, then email the receipt, with nothing happening after a failed charge.
  • Collapse a burst of model saves into one search-index rebuild per record.
  • Stop two jobs for the same account writing to the same rows at once.

More examples

Dispatching the batch and reporting on it

The callbacks are serialised and run later in a worker, so $this is not available inside them and neither is any object that cannot be serialised. That is why $importId is copied into a local variable first rather than captured through $import. Store $batch->id and a status endpoint can later read Bus::findBatch($id) for progress(), totalJobs, pendingJobs and failedJobs.

Example · php
<?php
// app/Http/Controllers/ImportController.php

declare(strict_types=1);

namespace App\Http\Controllers;

use App\Jobs\ImportCsvChunk;
use App\Models\Import;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;
use Throwable;

class ImportController extends Controller
{
    public function store(Import $import)
    {
        $importId = $import->id;

        $batch = Bus::batch(
            collect(range(1, $import->row_count))
                ->chunk(500)
                ->map(fn ($rows) => new ImportCsvChunk(
                    $importId,
                    $rows->first(),
                    $rows->last(),
                ))
                ->all()
        )
            ->name("import:{$importId}")
            ->allowFailures()
            ->then(fn (Batch $batch) => Import::find($importId)
                ->update(['status' => 'complete']))
            ->catch(fn (Batch $batch, Throwable $e) => logger()->error(
                'Import chunk failed', ['batch' => $batch->id]
            ))
            ->finally(fn (Batch $batch) => logger()->info(
                'Import finished', ['failed' => $batch->failedJobs]
            ))
            ->dispatch();

        return response()->json(['batch_id' => $batch->id]);
    }
}

A chain that stops at the first failure

If ChargeInvoice exhausts its attempts, the PDF and the receipt are never dispatched at all, which is what you want: no receipt for a payment that did not happen. Each job keeps its own tries and backoff, so a chain is slow to fail. Inside a chained job, prependToChain and appendToChain add further steps at run time.

Example · php
<?php

use App\Jobs\GenerateInvoicePdf;
use App\Jobs\ChargeInvoice;
use App\Jobs\SendReceipt;
use Illuminate\Support\Facades\Bus;
use Throwable;

$invoiceId = $invoice->id;

Bus::chain([
    new ChargeInvoice($invoice),
    new GenerateInvoicePdf($invoice),
    new SendReceipt($invoice),
])
    ->onQueue('billing')
    ->catch(function (Throwable $e) use ($invoiceId) {
        logger()->error('Billing chain failed', ['invoice' => $invoiceId]);
    })
    ->dispatch();

// `new ChargeInvoice($invoice)` and the parenthesis-free style the
// Laravel docs often use, `new SendReceipt`, are the same expression;
// PHP allows the parentheses to be dropped when there are no arguments.

One of these, not twelve

Save a product twelve times in a minute and twelve dispatches happen, but only one job is queued: the rest find the lock held and are discarded without error. uniqueId must identify the work, not the dispatch, which is why it is the product key. UniqueFor caps how long a lost lock can block new dispatches, so pick a value comfortably above the job's worst run time.

Example · php
<?php
// app/Jobs/RebuildProductIndex.php

declare(strict_types=1);

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\UniqueFor;
use Illuminate\Queue\Middleware\Skip;

#[UniqueFor(3600)]
class RebuildProductIndex implements ShouldQueue, ShouldBeUnique
{
    use Queueable;

    public function __construct(
        public int $productId,
        public bool $force = false,
    ) {}

    public function uniqueId(): string
    {
        return (string) $this->productId;
    }

    public function middleware(): array
    {
        // Shorthand: a plain boolean.
        return [Skip::when($this->force === false && app()->runningUnitTests())];

        // Longhand: the closure form. Skip::when() puts whatever you give
        // it through value(), so the closure is called straight away, not
        // deferred; both lines decide at the same moment.
        // return [Skip::when(fn (): bool => $this->force === false
        //     && app()->runningUnitTests())];
    }

    public function handle(): void
    {
        // ...
    }
}

Related

laravel jobslaravel queue reliabilitylaravel queue driverslaravel adv chunkinglaravel horizon

Discussion

  • Be the first to comment on this lesson.