📊 Change the order Laravel picks up queued jobs (AKA Priority Queues)
Tags: laravel, queue, php
Published: 2024-05-20 08:17:22

Have you ever wished some queued jobs would be processed way ahead others? For example, if you want your VIP client's transanctions to be completed before everyone else's, this article might help you.

## Simple approach ✨

If you have only a few tiers of clients, you might want to consider assigning them to a different queue so that VIP clients won't be blocked by other customers:

```php
<?php

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

class ProcessJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function __construct(private Customer $customer)
    {
        $this->onQueue($this->determineQueue());
    }

    /**
     * Determine which queue to use based on client type.
     */
    private function determineQueue(): string
    {
        return $customer->is_vip ? 'vip' : 'default';
    }

    public function handle(): void
    {
        // Job logic
    }
}
```

⚙️ Then running separate workers for both queues:

```bash
php artisan queue:work --queue=default
php artisan queue:work --queue=vip
```

✅ This is really easy and will work with any queue driver, like Redis, SQS, Database, etc.

❌ Unfortunately, this is not really scalable if you wish to have more control over the order of queued jobs or have many different tiers of users.

## Queue prioritization using custom queue driver 👨🏻‍💻

If you wish to control the order of transactions in more sophisticated way, for example based on a transaction amount or any other factor, consider creating a custom queue driver.

### 1. Add `priority` column to `jobs` table

To start, add a `priority` column in your `jobs` table. You can do this by creating a migration:

```bash
php artisan make:migration add_priority_to_jobs_table
```

Update the migration file:

```php
public function up()
{
    Schema::table('jobs', function (Blueprint $table) {
        $table->unsignedInteger('priority')->default(0); // Default priority is 0
    });
}

public function down()
{
    Schema::table('jobs', function (Blueprint $table) {
        $table->dropColumn('priority');
    });
}
```

Run the migration:

```bash
php artisan migrate
```

### 2. Create a Custom Queue Connector

Create a custom queue connector that extends Laravel's database queue connector. In this custom connector, you'll override the `push` and `pop` methods to handle job priority:

```php
<?php

namespace App\Queue;

use Illuminate\Queue\DatabaseQueue;
use Illuminate\Contracts\Queue\Job;
use Illuminate\Queue\Jobs\DatabaseJob;
use Illuminate\Queue\Jobs\DatabaseJobRecord;

class PriorityDatabaseQueue extends DatabaseQueue
{
    /**
     * Push a new job onto the queue.
     *
     * @param  string  $queue
     * @param  string  $payload
     * @param  string|null  $queue
     * @return mixed
     */
    public function push($job, $data = '', $queue = null)
    {
        $queue = $this->getQueue($queue);
        $payload = $this->createPayload($job, $queue, $data);

        return $this->database->table($this->table)->insertGetId([
            'queue' => $queue,
            'payload' => $payload,
            'attempts' => 0,
            'reserved' => 0,
            'reserved_at' => null,
            'available_at' => $this->availableAt(),
            'created_at' => $this->currentTime(),
            'priority' => $payload['data']['priority'] ?? 0, // Set the priority
        ]);
    }

    /**
     * Pop the next job off of the queue with prioritization.
     *
     * @param  string|null  $queue
     * @return \Illuminate\Contracts\Queue\Job|null
     */
    public function pop($queue = null)
    {
        $queue = $this->getQueue($queue);

        $job = $this->database->table($this->table)
            ->where('queue', $queue)
            ->where('reserved', 0)
            ->orderBy('priority', 'desc') // Prioritize jobs based on their priority level
            ->orderBy('id', 'asc')
            ->lockForUpdate()
            ->first();

        if (!is_null($job)) {
            return new DatabaseJob(
                $this->container, $this, new DatabaseJobRecord((object) $job)
            );
        }
    }
}
```

### 3. Create a Custom Connector Class


Create a custom connector class to register the custom queue connector:

```php
<?php

namespace App\Queue\Connectors;

use Illuminate\Queue\Connectors\DatabaseConnector;
use App\Queue\PriorityDatabaseQueue;

class PriorityDatabaseConnector extends DatabaseConnector
{
    public function connect(array $config)
    {
        return new PriorityDatabaseQueue(
            $this->connections->connection($config['connection'] ?? null),
            $config['table'],
            $config['queue'],
            $config['retry_after'] ?? 60
        );
    }
}
```

### 4. Register the Custom Connector

Register your custom connector in a service provider.

```php
use Illuminate\Support\ServiceProvider;
use App\Queue\Connectors\PriorityDatabaseConnector;

class AppServiceProvider extends ServiceProvider
{
    public function register(): void
    {
        $this->app->resolving('queue', function ($queue) {
            $queue->addConnector('prioritydatabase', function () {
                return new PriorityDatabaseConnector($this->app['db']);
            });
        });
    }
}
```

From now on, using a `prioritydatabase` driver will use our custom logic to push and fetch jobs to the queue based on their priority.

### 5. Configure Queue Connection 

Update your `config/queue.php` to use the new driver:

```php
'connections' => [
    'prioritydatabase' => [
        'driver' => 'prioritydatabase',
        'table' => 'jobs',
        'queue' => 'priority',
        'retry_after' => 60,
    ],
],
```

### 6. Set Job Priority When Dispatching Jobs

When dispatching jobs, set the priority level. You can do this by adding a `$priority` field to the job's payload.

```php
class SomeJob implements ShouldQueue
{
    public function __construct(public int $priority = 0)
    {
    }

    public function handle(): void
    {
        // Job logic
    }
}

// Dispatching the job with priority
$job = new SomeJob(10); // Priority 10
dispatch($job)->onQueue('priority');
```

With these steps, you will have a priority queue setup in Laravel where jobs are stored with a `priority` column in the `jobs` table. The custom queue connector ensures that jobs are popped from the queue based on their priority, allowing higher priority jobs to be processed first.

Feel free to customize the way you pop the jobs based on your needs.

✅ More control over the job processing order.
❌ Changing the queue driver requires major refactor to adapt the popping logic. Not all drivers might be supported.

## Summary

🕵🏻‍♂️ Digging deep in Laravel queue connectors could be fun and it's interesting to learn how it works internally so that we are able to customize it to our needs.

**While the DB driver is easy to understand and customize, I strongly suggest not to use database driver in production**. If you wish to maintain scalability and avoid lock issues, I recommend creating a similar connector based on Redis with `ZADD` and `ZRANGE/ZREM` to handle similar logic instead.