Implementing a Redis Message Queue with RPUSH and BLPOP in PHP

Background

In high‑concurrency business systems, a message queue is often used to solve concurrency problems by processing tasks asynchronously. This article introduces one Redis message‑queue solution that combines RPUSH and BLPOP, using a cancel‑order scenario (the business logic is omitted, focusing on the queue implementation).

Solution Overview

Data is inserted at the tail of a Redis list with the RPUSH command, and BLPOP is used to block‑wait and retrieve items from the head of the list (FIFO). The blocking retrieval is placed in backend code that consumes the queue. The consumer function consume(app/WorkQueue.php->consume()) is triggered by a scheduled task every 5 minutes; the script automatically exits after 290 seconds so that the next scheduled process can start.

Implementation

The configuration file app/config/config.php defines queue drivers; for the cancel‑order driver, the class value points to app/logic/Order.php. The method specifies the business‑logic method to execute, and the open flag enables or disables the queue.

The consumer is invoked via a cron entry: */5 * * * * php /app/WorkQueue.php cancel_order The main consumer script ( app/WorkQueue.php) performs the following steps:

<?php
require_once './config/config.php';

class WorkQueue {
    /**
     * Execute message‑queue consumption
     */
    public function consume() {
        $startTime = time(); // start time
        $allowAllExcuteTime = 290; // max execution time before graceful exit
        $configArr = include_once './config/config.php';
        $quueDriverConfig = $configArr['quueDriver'];
        $driverName = $argv[1]; // e.g., cancel_order
        $redisQueueKey = "QUEUE_{$driverName}"; // queue key prefix
        $logicName = $quueDriverConfig[$driverName]['class'];
        $method = $quueDriverConfig[$driverName]['method'];
        if ($quueDriverConfig[$driverName]['open'] !== true) {
            return;
        }
        require_once "./logic/queue/{$logicName}.php";
        $logicQueueObj = new $logicName();
        $redis = new redis();
        $redis->connect('127.0.0.1', 6379);
        while (true) {
            $specialTime = $allowAllExcuteTime - (time() - $startTime);
            if ($specialTime <= 0) {
                break;
            }
            $content = $redis->BLPOP($redisQueueKey, $specialTime);
            $contentArr = json_decode($content, true);
            if (is_array($content)) {
                $logicQueueObj->$method($contentArr);
            }
        }
    }
}
$workQueueObj = new WorkQueue();
$workQueueObj->consume();

BLPOP blocks on the queue key $redisQueueKey for up to $specialTime seconds; if no data is present, it waits until the timeout. When data is retrieved, it is decoded from JSON and passed to the configured business‑logic method.

Configuration

The configuration file defines each queue driver under quueDriver. For the cancel_order driver, the class points to the order‑handling class, and method indicates which method to invoke. Setting open to true enables the queue; otherwise the consumer exits immediately.

Redis Related

Queue keys use the prefix QUEUE_. Queue items are stored as JSON strings; during consumption they are decoded into associative arrays.

Executing Queue Consumption

The consumer runs via the cron command shown above. It connects to Redis, repeatedly calls BLPOP with a dynamically calculated timeout, and processes each task using the specified logic class and method.

Conclusion

There are many ways to implement a queue; the key is to choose the approach that fits your business scenario. In real‑world usage you may also need to handle cases such as re‑queueing failed tasks, which is beyond the scope of this brief guide.