PHP源码队列系统实现_PHP源码队列系统实现指南

CREATE TABLE `jobs` (
    `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    `payload` JSON NOT NULL COMMENT '任务内容，JSON格式，包含任务类型和参数',
    `status` ENUM('pending', 'processing', 'failed', 'completed') NOT NULL DEFAULT 'pending' COMMENT '任务状态',
    `attempts` TINYINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '尝试次数',
    `available_at` DATETIME NOT NULL COMMENT '任务可执行时间，用于延迟任务或重试',
    `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX `idx_status_available` (`status`, `available_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

payload

{'type': 'send_email', 'user_id': 123, 'subject': 'Welcome!'}

available_at

jobs

<?php
// producer.php
function dispatchJob(string $type, array $data, int $delaySeconds = 0) {
    $pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'password');
    $stmt = $pdo->prepare("INSERT INTO jobs (payload, status, available_at) VALUES (?, ?, ?)");

    $payload = json_encode(['type' => $type, 'data' => $data]);
    $availableAt = date('Y-m-d H:i:s', time() + $delaySeconds);

    $stmt->execute([$payload, 'pending', $availableAt]);
    echo "任务 [{$type}] 已入队。\n";
}

// 示例：发送欢迎邮件
dispatchJob('send_welcome_email', ['user_id' => 456, 'email' => 'test@example.com']);

// 示例：延迟1分钟生成报告
dispatchJob('generate_report', ['report_id' => 789], 60);

jobs

pending

available_at

<?php
// worker.php
require 'JobProcessor.php'; // 假设你有一个处理不同任务类型的类

function startWorker() {
    $pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'password');
    $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

    echo "Worker 启动，开始监听任务...\n";

    while (true) {
        $pdo->beginTransaction();
        try {
            // 尝试获取一个可用的任务并加锁
            $stmt = $pdo->prepare("
                SELECT * FROM jobs
                WHERE status = 'pending' AND available_at <= NOW()
                ORDER BY id ASC
                LIMIT 1
                FOR UPDATE
            ");
            $stmt->execute();
            $job = $stmt->fetch(PDO::FETCH_ASSOC);

            if ($job) {
                // 标记任务为处理中
                $updateStmt = $pdo->prepare("UPDATE jobs SET status = 'processing', attempts = attempts + 1, updated_at = NOW() WHERE id = ?");
                $updateStmt->execute([$job['id']]);
                $pdo->commit(); // 提交事务，释放锁，让其他worker可以继续拉取

                echo "正在处理任务 #{$job['id']}...\n";
                $payload = json_decode($job['payload'], true);
                $jobType = $payload['type'];
                $jobData = $payload['data'];

                try {
                    // 实际执行任务
                    JobProcessor::process($jobType, $jobData);

                    // 任务成功，标记为完成
                    $successPdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'password'); // 新PDO连接避免事务冲突
                    $successStmt = $successPdo->prepare("UPDATE jobs SET status = 'completed', updated_at = NOW() WHERE id = ?");
                    $successStmt->execute([$job['id']]);
                    echo "任务 #{$job['id']} [{$jobType}] 完成。\n";

                } catch (Throwable $e) {
                    // 任务失败，标记为失败或重试
                    $failPdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'password');
                    $failStmt = $failPdo->prepare("UPDATE jobs SET status = 'failed', updated_at = NOW() WHERE id = ?, error_message = ?"); // 增加error_message字段
                    $failStmt->execute([$job['id'], $e->getMessage()]);
                    echo "任务 #{$job['id']} [{$jobType}] 失败: {$e->getMessage()}\n";
                    // 这里可以根据attempts字段实现重试逻辑，比如更新available_at为未来某个时间
                }
            } else {
                $pdo->commit(); // 没有任务，也要提交事务
                // echo "没有待处理任务，等待...\n";
                sleep(1); // 没有任务时等待1秒，避免CPU空转
            }
        } catch (Throwable $e) {
            $pdo->rollBack();
            echo "数据库操作异常: " . $e->getMessage() . "\n";
            sleep(5); // 出现异常时等待一段时间再重试
        }
    }
}

// 简单的任务处理器示例
class JobProcessor {
    public static function process(string $type, array $data) {
        switch ($type) {
            case 'send_welcome_email':
                echo "发送欢迎邮件给用户 {$data['user_id']} ({$data['email']})...\n";
                // 模拟耗时操作
                sleep(rand(1, 3));
                // if (rand(0, 10) < 2) throw new Exception("模拟邮件发送失败"); // 模拟失败
                break;
            case 'generate_report':
                echo "生成报告 {$data['report_id']}...\n";
                sleep(rand(2, 5));
                break;
            default:
                throw new Exception("未知任务类型: {$type}");
        }
    }
}

startWorker();

worker.php

Supervisor

systemd

jobs

attempts

max_attempts

attempts

worker.php

catch

// ... 在任务失败的catch块中
$maxAttempts = 3; // 假设最大重试3次

$currentAttempts = (int)$job['attempts']; // 获取当前尝试次数
if ($currentAttempts < $maxAttempts) {
    // 标记为pending，并设置available_at为未来某个时间，实现延迟重试
    $retryDelaySeconds = pow(2, $currentAttempts) * 60; // 简单的指数退避策略，比如1分钟，2分钟，4分钟...
    $retryAvailableAt = date('Y-m-d H:i:s', time() + $retryDelaySeconds);

    $retryStmt = $failPdo->prepare("UPDATE jobs SET status = 'pending', available_at = ?, attempts = ?, error_message = ? WHERE id = ?");
    $retryStmt->execute([$retryAvailableAt, $currentAttempts + 1, $e->getMessage(), $job['id']]);
    echo "任务 #{$job['id']} [{$jobType}] 失败，将在 {$retryDelaySeconds} 秒后重试。\n";
} else {
    // 超过最大重试次数，标记为永久失败
    $failStmt = $failPdo->prepare("UPDATE jobs SET status = 'failed', updated_at = NOW(), error_message = ? WHERE id = ?");
    $failStmt->execute([$e->getMessage() . " (达到最大重试次数)", $job['id']]);
    echo "任务 #{$job['id']} [{$jobType}] 达到最大重试次数，标记为永久失败。\n";
}

jobs

dlq_reason

payload

failed_jobs

payload

pending

failed

failed

pending

if/else