Why a queue
Some work shouldn't happen while someone is waiting for a page: sending emails, delivering webhooks, resizing uploaded images, checking that a payment went through, calling a slow API. The page records what needs doing and returns straight away, and something in the background does the work.
In a PHP application the things that need doing are usually rows in a table already. An email that hasn't been sent yet. An upload that hasn't been resized. A payment that hasn't been checked. So the question is how to work through those rows reliably, and with more than one worker when there's a lot to do.
The setup
Let's use emails. The table records each email, whether it's been sent, and how many times sending has been tried:
CREATE TABLE email (
id BIGINT NOT NULL AUTO_INCREMENT,
recipient VARCHAR(250) NOT NULL,
subject VARCHAR(250) NOT NULL,
body TEXT NOT NULL,
status ENUM('waiting', 'sent', 'failed') NOT NULL DEFAULT 'waiting',
attempts INT NOT NULL DEFAULT 0,
retry_at BIGINT NULL,
PRIMARY KEY (id),
INDEX (status, retry_at)
);A page that wants to send an email just inserts a row, and carries on.
The simple way
The obvious worker is a script that cron runs every minute. It picks up the emails that are waiting, sends them, and marks each one sent:
$emails = $db->query("
SELECT * FROM email WHERE status = 'waiting' ORDER BY id LIMIT 100
")->fetchAll(PDO::FETCH_OBJ);
$markSent = $db->prepare("UPDATE email SET status = 'sent' WHERE id = ?");
foreach ($emails as $email) {
sendEmail($email);
$markSent->execute([$email->id]);
}What goes wrong
As long as there's only ever one of these running, it works. But:
- If a batch takes longer than a minute, cron starts the next run while the first is still going. Both select the same waiting rows, and people get two copies of the same email.
- For the same reason you can't run more workers to get through a backlog faster.
- If one email fails every time, it's tried again every minute forever, at the front of the queue.
- If the script dies half way through, nothing tells you which emails in the batch went out.
The usual fix is to claim rows before working on them. You add some columns to the table (who has claimed the row, when the claim runs out, how many tries it's had), or a separate jobs table with a status of pending, running or done. A worker claims a batch with an UPDATE, then reads back the rows it claimed:
UPDATE email SET claimed_by = :worker, claim_expires = :expires
WHERE claimed_by IS NULL AND status = 'waiting'
LIMIT 10;
SELECT * FROM email WHERE claimed_by = :worker;That stops the duplicates, but now every table that's a queue carries bookkeeping columns and indexes that have nothing to do with what the table is for. A worker that crashes leaves its rows claimed, so claims have to expire and something has to notice and release them. And claiming and releasing are extra writes for every item. The database already knows how to do all of this.
Let MySQL do the locking
MySQL 8 has SELECT ... FOR UPDATE SKIP LOCKED. FOR UPDATE locks the rows a query returns until the end of the transaction. SKIP LOCKED says that if a row is already locked by someone else, don't wait for it: leave it out and carry on to the next. So two workers running the same query at the same moment get different rows.
$db->beginTransaction();
$id = $db->query("
SELECT id FROM email
WHERE status = 'waiting'
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED
")->fetchColumn();
if ($id !== false) {
$q = $db->prepare("SELECT * FROM email WHERE id = ?");
$q->execute([$id]);
sendEmail($q->fetch(PDO::FETCH_OBJ));
$db->prepare("UPDATE email SET status = 'sent' WHERE id = ?")
->execute([$id]);
}
$db->commit();While the transaction is open, the row is locked and every other worker's query skips over it. Sending marks it sent, so once the transaction commits the row doesn't match the query any more: it has left the queue.
If sending throws, we roll back instead. The status stays at waiting, the lock is released, and the row is back in the queue for the next worker. If the worker dies part way through (killed, out of memory, the server restarting), its connection to MySQL drops, MySQL rolls the transaction back, and again the row is simply back in the queue. There are no claims to expire and nothing to clean up.
The queue is the query
Notice what we didn't need: a queue table, or any columns for the queue. A row that matches the query is, by definition, in the queue. Processing it changes something about the row that takes it out of the query, and that's how it leaves.
So any table can be a queue without being changed at all, as long as there's a query that picks out the rows that need work. A table can be several queues at once, with a different query for each. ORDER BY sets the priority, and a condition on a time column turns it into a schedule (AND send_after <= :now).
The one rule is that processing an item must change the row so that it no longer matches the query. If it doesn't, the row comes straight back round.
A small library
Each queue is then just two functions. One locks the next items and returns their ids, using FOR UPDATE SKIP LOCKED; it takes a limit so the same query can be used for counting. The other processes one item:
function lockNextEmails(PDO $db, int $limit): array {
$q = $db->prepare("
SELECT id FROM email
WHERE status = 'waiting'
AND (retry_at IS NULL OR retry_at <= :now)
ORDER BY id
LIMIT :limit
FOR UPDATE SKIP LOCKED
");
$q->bindValue(':now', time(), PDO::PARAM_INT);
$q->bindValue(':limit', $limit, PDO::PARAM_INT);
$q->execute();
return array_map('intval', $q->fetchAll(PDO::FETCH_COLUMN));
}
function processEmail(PDO $db, int $id): void {
$q = $db->prepare("SELECT * FROM email WHERE id = ?");
$q->execute([$id]);
$email = $q->fetch(PDO::FETCH_OBJ);
sendEmail($email);
$db->prepare("UPDATE email SET status = 'sent' WHERE id = ?")
->execute([$id]);
}Everything else is the same for every queue. Processing one item is a transaction around those two calls:
function processNext(PDO $db, callable $lockNext, callable $process): bool {
$db->beginTransaction();
try {
$ids = $lockNext($db, 1);
if (!$ids) {
$db->commit();
return false; // nothing waiting
}
$process($db, $ids[0]);
$db->commit();
return true;
} catch (Throwable $e) {
$db->rollBack(); // the item goes back into the queue
throw $e;
}
}A worker is a loop that processes items until there are none, then waits a few seconds and looks again:
function runWorker(PDO $db, callable $lockNext, callable $process, int $idle = 5): void {
while (true) {
try {
if (!processNext($db, $lockNext, $process)) {
sleep($idle);
}
} catch (Throwable $e) {
error_log($e->getMessage());
sleep(1);
}
}
}
runWorker($db, 'lockNextEmails', 'processEmail');Run as many copies of that as you like, on as many machines as you like, under supervisor or systemd or one per container. They don't know about each other and don't need to; MySQL hands each of them a different row. How fast a queue goes is just how many workers it has.
Counting what's waiting uses the same query with a bigger limit, in a transaction that commits straight away so the locks are let go at once:
function countWaiting(PDO $db, callable $lockNext): int {
$db->beginTransaction();
$ids = $lockNext($db, 100);
$db->commit();
return count($ids);
}It stops at 100 so it stays quick on a big backlog, and it only counts rows nobody is working on right now. That's enough for a dashboard, or for deciding whether to start more workers.
Failures and retries
Throwing an exception puts the item straight back in the queue, which is right for a passing problem like a timeout. It's wrong for an email that will never send: it would come straight back round, again and again, and with ORDER BY id it would stay at the front.
For failures that might not go away, the process function records the failure on the row instead of throwing, and that takes the row out of the query for a while, or for good. The queue still doesn't need anything of its own: how many times an email has been tried and when to try it again are facts about the email.
function processEmail(PDO $db, int $id): void {
$q = $db->prepare("SELECT * FROM email WHERE id = ?");
$q->execute([$id]);
$email = $q->fetch(PDO::FETCH_OBJ);
try {
sendEmail($email);
$db->prepare("UPDATE email SET status = 'sent' WHERE id = ?")
->execute([$id]);
} catch (MailError $e) {
$attempts = $email->attempts + 1;
if ($attempts >= 5) {
// give up: 'failed' is out of the queue for good
$db->prepare("UPDATE email SET status = 'failed', attempts = ?
WHERE id = ?")
->execute([$attempts, $id]);
} else {
// try again later: a minute, then two, then three...
$db->prepare("UPDATE email SET attempts = ?, retry_at = ? WHERE id = ?")
->execute([$attempts, time() + 60 * $attempts, $id]);
}
}
}Because the update happens inside the same transaction, it commits along with everything else, and the row leaves the query until its retry_at comes round.
Things to know
- The transaction stays open while an item is processed, holding a database connection and the row's lock. A few seconds is fine. For work that takes minutes, like rendering video on a GPU, it's better for a worker to claim the item and report its progress, which is a different design.
- Sending an email can't be rolled back. If a worker dies after the email has gone but before the commit, the row goes back into the queue and the email is sent again. Like nearly every queue, this one delivers at least once, so where it matters, make processing safe to repeat.
- Keep the queue's query on an index (here
status, retry_at), so finding the next row is quick and doesn't touch rows it doesn't need. SKIP LOCKEDneeds MySQL 8.0 or MariaDB 10.6 or later, and InnoDB tables.
That's all there is to it: a query that says what's waiting, a function that does the work and changes the row, and MySQL's own locks sharing it all out.