PHP任务队列：实现与优化指南

在 PHP 中处理任务队列：从简单到复杂

在现代 Web 应用程序中，经常需要执行资源密集型或长时间运行的操作，而无需让用户等待。发送电子邮件通知、生成报告、处理图像、与外部 API 集成——所有这些任务都非常适合异步执行。这时任务队列就派上了用场。在本文中，我们将详细介绍如何在 PHP 中实现和优化队列处理，从简单的解决方案开始，一直到工业级工具。

什么是任务队列及其必要性？

任务队列是一种允许将某些工作推迟到以后执行的机制。您不是立即在 HTTP 请求中执行任务，而是将其放入队列中。一个或多个专门的工作进程从队列中取出任务并在后台处理它们。

使用队列的主要优势：

• 对用户响应更快：用户立即收到响应，而繁重的工作则异步执行。
• 可扩展性：您可以运行多个工作进程来并行处理任务。
• 可靠性：如果任务失败（例如，由于临时错误），可以在以后重试。
• 减轻主服务器负担：后台进程不会阻塞新 HTTP 请求的处理。

基于数据库的简单队列

在 PHP 中组织队列的最简单方法是使用数据库表（例如 MySQL）。这种方法不需要安装额外的软件，非常适合小型项目。

第 1 步：创建 jobs 表

CREATE TABLE jobs (    id INT AUTO_INCREMENT PRIMARY KEY,    queue_name VARCHAR(255) DEFAULT 'default',    payload TEXT NOT NULL,    status ENUM('pending', 'processing', 'completed', 'failed') DEFAULT 'pending',    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,    available_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,    attempts INT DEFAULT 0);

第 2 步：向队列添加任务

function addJob(string $queueName, array $data): void {    $pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'password');    $stmt = $pdo->prepare(        "INSERT INTO jobs (queue_name, payload, available_at)          VALUES (?, ?, NOW())"    );    $stmt->execute([$queueName, json_encode($data)]);}

// Example usageaddJob('emails', ['to' => 'user@example.com', 'subject' => 'Hello!']);

第 3 步：创建工作进程 (worker.php)

// Infinite loop for processing jobswhile (true) {    $pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'password');        // Capture a job with row locking    $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) {        // Mark as "processing"        $updateStmt = $pdo->prepare("UPDATE jobs SET status = 'processing' WHERE id = ?");        $updateStmt->execute([$job['id']]);                try {            // Execute the job            $data = json_decode($job['payload'], true);            processJob($data);                        // Mark as completed            $pdo->prepare("UPDATE jobs SET status = 'completed' WHERE id = ?")                ->execute([$job['id']]);        } catch (Exception $e) {            // Error handling            $attempts = $job['attempts'] + 1;            if ($attempts < 3) {