PHP 佇列實作指南：從原理到生產級類庫封裝

佇列核心概念

佇列 是先進先出 (FIFO) 的線性資料結構，僅允許尾端插入、頭端移除。在 Web 開發中，佇列常用於：

• 削峰填谷：高併發請求寫入佇列，背景消費者平滑處理
• 非同步解耦：訂單建立後發送通知、更新搜尋索引等非即時任務
• 任務排程：定時任務、重試機制、優先級處理
• 減輕 DB 壓力：合併寫入 (如計數器 +5 取代 5 次 +1)

陣列實作簡易佇列

<?php
class SimpleQueue {
    private $data = [];
    
    public function enqueue($value) { $this->data[] = $value; }
    public function dequeue() { return array_shift($this->data); }
    public function peek() { return $this->data[0] ?? null; }
    public function size() { return count($this->data); }
    public function isEmpty() { return empty($this->data); }
}

生產級緩存支援佇列類 (支援多進程)

<?php
/**
 * 基於緩存驅動的佇列實作
 * 支援: Redis, Memcached, APCu, File 等實現 Cache 接口的驅動
 */
class Queue {
    private $_queue = [];
    protected $cache;
    protected $cacheKey;
    
    public function __construct(string $queueName, $cacheDriver = null) {
        $this->cacheKey = 'queue_' . $queueName;
        $this->cache = $cacheDriver ?? Cache::instance(); // 需自行實作 Cache 單例
        
        $cached = $this->cache->get($this->cacheKey);
        if (is_array($cached)) $this->_queue = $cached;
    }
    
    /** 入佇列 */
    public function enQueue($value): self {
        $this->_queue[] = $value;
        $this->persist();
        return $this;
    }
    
    /** 批量出佇列 (預設 1 筆) */
    public function sliceQueue(int $num = 1): array {
        $num = min($num, count($this->_queue));
        $output = array_splice($this->_queue, 0, $num);
        $this->persist();
        return $output;
    }
    
    /** 單筆出佇列 */
    public function deQueue() {
        $entry = array_shift($this->_queue);
        $this->persist();
        return $entry;
    }
    
    /** 查看隊首 */
    public function peek() { return $this->_queue[0] ?? null; }
    
    /** 查看前 N 筆 */
    public function peeks(int $num): array {
        return array_slice($this->_queue, 0, min($num, count($this->_queue)));
    }
    
    /** 佇列長度 */
    public function size(): int { return count($this->_queue); }
    
    /** 銷毀佇列 */
    public function destroy(): void { $this->cache->remove($this->cacheKey); }
    
    private function persist(): void {
        $this->cache->set($this->cacheKey, $this->_queue);
    }
}

Redis 高效能佇列實作 (推薦生產環境)

<?php
class RedisQueue {
    private $redis;
    private $key;
    
    public function __construct(\Redis $redis, string $queueName) {
        $this->redis = $redis;
        $this->key = "queue:{$queueName}";
    }
    
    // LPUSH + RPOP = FIFO
    public function push($data): int {
        return $this->redis->lPush($this->key, json_encode($data));
    }
    
    public function pop(): ?array {
        $raw = $this->redis->rPop($this->key);
        return $raw ? json_decode($raw, true) : null;
    }
    
    // 阻塞式 pop (長輪詢，適合消費者)
    public function bPop(int $timeout = 5): ?array {
        $raw = $this->redis->bRPop($this->key, $timeout);
        return $raw ? json_decode($raw[1], true) : null;
    }
    
    public function size(): int { return $this->redis->lLen($this->key); }
    public function clear(): bool { return $this->redis->del($this->key) > 0; }
}

實戰應用場景：訪問計數器合併寫入

<?php
// 前端請求：快速寫入佇列 (毫秒級響應)
$queue = new RedisQueue($redis, 'page_views');
$queue->push(['page_id' => 123, 'time' => time()]);
echo "OK";

// 背景 Worker：批次處理 (每分鐘/每 1000 筆)
$batch = [];
while (($item = $queue->pop()) && count($batch) < 1000) {
    $batch[] = $item;
}
if ($batch) {
    // 聚合統計
    $counts = [];
    foreach ($batch as $item) {
        $counts[$item['page_id']] = ($counts[$item['page_id']] ?? 0) + 1;
    }
    // 批次更新 DB
    foreach ($counts as $pageId => $cnt) {
        $pdo->prepare("UPDATE pages SET views = views + ? WHERE id = ?")
            ->execute([$cnt, $pageId]);
    }
}

佇列選型建議

最佳實踐總結

1. 選對工具：小專案用 Redis List，大系統上 RabbitMQ/Kafka。
2. 冪等設計：消費者邏輯必須冪等，配合唯一 ID 去重。
3. 監控告警：監控佇列長度、消費延遲、失敗率。
4. 死信佇列：重試 N 次失敗移入 DLQ，人工介入。
5. 優先級：重要任務用獨立高優先級佇列。

佇列是高併發架構的核心組件，合理引入能大幅提升系統彈性與響應速度，建議從 Redis List 起步，隨業務增長平滑遷移。