PHP 多進程 Job Queue 處理器：pcntl_fork 實作佇列消費、狀態追蹤與錯誤重試

架構概覽

本範例展示一個 CLI 專用的 Job Queue 消費者：從資料庫撈取待處理任務，pcntl_fork 產生子進程執行，父進程 pcntl_waitpid 等待並根據退出碼更新狀態（成功/失敗/執行中）。

核心類別結構

<?php
declare(ticks=1); // 訊號處理需要

class JobQueueWorker
{
    private PDO $pdo;
    private int $maxJobsPerRun = 10;
    private int $idleSleep = 5;
    private int $jobCount = 0;
    
    public function __construct(PDO $pdo)
    {
        $this->pdo = $pdo;
        $this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
    }
    
    public function run(): void
    {
        // 僅允許 CLI 執行
        if (PHP_SAPI !== 'cli') {
            die("Only CLI allowed\n");
        }
        
        // 訊號處理：優雅關閉
        pcntl_signal(SIGTERM, [$this, 'shutdown']);
        pcntl_signal(SIGINT, [$this, 'shutdown']);
        
        echo "Job Worker 啟動 (PID: " . getmypid() . ")\n";
        
        while (!$this->shouldStop) {
            $this->processBatch();
        }
        
        echo "Job Worker 正常關閉\n";
    }
    
    private bool $shouldStop = false;
    public function shutdown(): void
    {
        echo "收到關閉訊號，完成當前批次後退出...\n";
        $this->shouldStop = true;
    }
    
    private function processBatch(): void
    {
        // 1. 撈取待處理任務（狀態 1=待處理, 2=執行中）
        $stmt = $this->pdo->query("
            SELECT id, job_uri, paras 
            FROM job_queue 
            WHERE status = 1 AND approved = 1 
            ORDER BY id 
            LIMIT 20
        ");
        $jobs = $stmt->fetchAll();
        
        if (empty($jobs)) {
            if (++$this->jobCount >= $this->maxJobsPerRun) {
                echo "連續 {$this->maxJobsPerRun} 次無任務，休眠較久...\n";
                sleep(30);
                $this->jobCount = 0;
            } else {
                sleep($this->idleSleep);
            }
            return;
        }
        
        $this->jobCount = 0; // 重置空閒計數
        
        foreach ($jobs as $job) {
            if ($this->shouldStop) break;
            
            $this->executeJob($job);
        }
    }
    
    private function executeJob(array $job): void
    {
        $jobId = $job['id'];
        
        // 2. 標記為執行中（防止重複消費）
        $this->pdo->prepare("UPDATE job_queue SET status=2, started_at=NOW() WHERE id=?")
                  ->execute([$jobId]);
        
        // 3. Fork 子進程
        $pid = pcntl_fork();
        
        if ($pid === -1) {
            // Fork 失敗
            $this->markFailed($jobId, 'fork_failed');
            return;
        }
        
        if ($pid === 0) {
            // ===== 子進程 =====
            $this->runJobLogic($job);
            exit(0); // 成功退出碼
        }
        
        // ===== 父進程 =====
        echo "等待 Job #{$jobId} (PID: $pid)...\n";
        
        // 等待子進程結束
        $status = 0;
        $waitPid = pcntl_waitpid($pid, $status, WUNTRACED);
        
        if ($waitPid === -1) {
            $this->markFailed($jobId, 'waitpid_failed');
            return;
        }
        
        // 解析退出狀態
        $exitCode = pcntl_wifexited($status) ? pcntl_wexitstatus($status) : -1;
        $signal   = pcntl_wifsignaled($status) ? pcntl_wtermsig($status) : 0;
        
        if ($exitCode === 0) {
            $this->markSuccess($jobId);
            echo "Job #{$jobId} 成功完成\n";
        } else {
            $reason = $signal ? "signal_$signal" : "exit_$exitCode";
            $this->markFailed($jobId, $reason);
            echo "Job #{$jobId} 失敗: $reason\n";
        }
    }
    
    private function runJobLogic(array $job): void
    {
        // 實際業務邏輯：這裡示範呼叫另一個 PHP 腳本
        $phpExec = '/usr/bin/php';
        $script  = '/var/www/app/index.php';
        $uri     = $job['job_uri'];
        $params  = json_decode($job['paras'] ?? '{}', true);
        
        $cmd = [$phpExec, $script, "--uri=$uri"];
        foreach ($params as $k => $v) {
            $cmd[] = "--$k=$v";
        }
        
        // 執行外部命令
        $descriptors = [
            0 => ['pipe', 'r'],
            1 => ['pipe', 'w'],
            2 => ['pipe', 'w'],
        ];
        $proc = proc_open($cmd, $descriptors, $pipes);
        
        if (!is_resource($proc)) {
            exit(1);
        }
        
        // 讀取輸出（避免管道阻塞）
        $stdout = stream_get_contents($pipes[1]);
        $stderr = stream_get_contents($pipes[2]);
        fclose($pipes[0]); fclose($pipes[1]); fclose($pipes[2]);
        
        $ret = proc_close($proc);
        
        if ($ret !== 0) {
            error_log("Job #{$job['id']} stderr: $stderr");
            exit($ret);
        }
    }
    
    private function markSuccess(int $jobId): void
    {
        $this->pdo->prepare("UPDATE job_queue SET status=999, finished_at=NOW() WHERE id=?")
                  ->execute([$jobId]);
    }
    
    private function markFailed(int $jobId, string $reason): void
    {
        $this->pdo->prepare("UPDATE job_queue SET status=-1, error_msg=?, finished_at=NOW() WHERE id=?")
                  ->execute([$reason, $jobId]);
    }
}

// === 啟動入口 ===
$pdo = new PDO('mysql:host=localhost;dbname=app;charset=utf8mb4', 'user', 'pass');
$worker = new JobQueueWorker($pdo);
$worker->run();

資料表結構建議

CREATE TABLE job_queue (
    id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    job_uri     VARCHAR(255) NOT NULL COMMENT '路由 URI',
    paras       JSON         NOT NULL COMMENT '參數 JSON',
    status      TINYINT      NOT NULL DEFAULT 1 COMMENT '1=待處理 2=執行中 999=成功 -1=失敗',
    approved    TINYINT      NOT NULL DEFAULT 1 COMMENT '審核通過才執行',
    error_msg   VARCHAR(500) DEFAULT NULL,
    created_at  TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP,
    started_at  TIMESTAMP    NULL DEFAULT NULL,
    finished_at TIMESTAMP    NULL DEFAULT NULL,
    INDEX idx_status_approved (status, approved)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

關鍵技術點解析

生產環境強化建議

• 分散式鎖：多 Worker 時用 Redis SETNX 或資料庫 SELECT ... FOR UPDATE SKIP LOCKED 避免重複消費。
• 超時機制：父進程 pcntl_alarm() 或非阻塞輪詢 pcntl_waitpid($pid, $status, WNOHANG) 實作任務逾時殺掉子進程。
• 重試策略：status=-1 任務寫入 retry_count，排程器定期將 retry_count < 3 重置為 status=1。
• 監控指標：佇列堆積數、處理耗時分位、成功率、Worker 存活數，接入 Prometheus/Grafana。
• 優雅部署：收到 SIGTERM 先停止接新任務，完成當前批次再退出，Supervisor stopwaitsecs 給足緩衝時間。

現代替代方案

若專案允許引入依賴，強烈建議使用成熟佇列系統：

• Laravel Horizon + Redis：Web UI、自動擴縮容、失敗重試、標籤分組
• Symfony Messenger：Transport 抽象、支援 RabbitMQ/Redis/DB/AMQP
• Swoole Task Worker：協程非阻塞、單機萬級併發、內建 Timer
• RabbitMQ / Kafka + PHP Consumer：企業級訊息佇列、持久化、ACK 機制、死信佇列

總結

pcntl_fork 實作 Job Queue 是理解多進程模型的好教材，但生產環境面臨訊號處理、殭屍進程、資源洩漏、可觀測性等挑戰。中小專案優先用 crontab + 單進程 CLI 或 框架內建排程；高併發、高可靠需求請直接上 Redis/RabbitMQ + 成熟 Consumer 框架。