Using a RDBM as Queue database would be a poor choice for concurrent job. It would cause your DB hand up with thoundsands of connections. Inspite of restarting, It’s not respond until you stop the queue servers that connect to it.

In this post I will help you build a simple, robust Queue worker using database. Few months ago, my boss called me at 4 a.m and said that the system was down and please do something. I quickly opened the AWS RDS dashboard and saw the strange thing. There are tons of database connections but the CPU usage percentage is just few percent. What was that ? It was really made me headache for a month. I made a question in stackoverflow and laracast but still no hope.

Then I deep dive in laravel core framework and find out this piece of code.

/**
* Get the next available job for the queue.
*
* @param  string|null  $queue
* @return \Illuminate\Queue\Jobs\DatabaseJobRecord|null
*/
protected function getNextAvailableJob($queue)
{
    $job = $this->database->table($this->table)
                ->lockForUpdate()
                ->where('queue', $this->getQueue($queue))
                ->where(function ($query) {
                    $this->isAvailable($query);
                    $this->isReservedButExpired($query);
                })
                ->orderBy('id', 'asc')
                ->first();
 
    return $job ? new DatabaseJobRecord((object) $job) : null;
}

Let’s see it more detail using pg_stat_statements

I will explain this snippet : it will pop a job from queue and delivery it to worker service and lock (lockForUpdate()) it while the worker service processing. Some reasons, service worker ‘s not respond, the current job is locked while another worker services begin transaction and wait for executing the next jobs. The result is the connection count grow up dramatically to few thousands.

You even cannot connect to database because there isn’t any slot. I will reproduce this issues as below:

Video

Since PostgreSQL 9.4 release FOR UPDATE … SKIP LOCKED , now I can FIX and rebuild concurrent work queues.

$job = DB::select('SELECT 
*
FROM jobs AS j
WHERE j.queue = :queue_name
AND j.reserved_at IS NULL
AND j.available_at <= extract(epoch from now())
OR j.reserved_at <= extract(epoch from now()) - :retry_after_second
ORDER BY j.id ASC
LIMIT 1
FOR UPDATE OF j SKIP LOCKED;', ['queue_name' => $this->getQueue($queue), 'retry_after_second' => $this->retryAfter]);
$job = reset($job);

Now, Worker services can get the jobs independently in order and the connection count now is equal workers count and no more connection explosion (y).

And well, That's it . If you have any questions, you can mail me. Thanks for reading!