PHP 任务队列：完整实战指南

PHP 任务队列：完整实战指南

在现代 Web 应用中，经常需要执行耗时或资源密集型操作，而无需让用户等待。发送邮件、生成 PDF 报告、处理图像、数据抓取——所有这些任务都非常适合通过任务队列进行异步处理。在本文中，我们将详细讲解如何在实际中组织 PHP 任务队列，介绍常用工具并编写可运行的代码。

什么是任务队列，为什么需要它？

任务队列（job queue）是一种机制，允许将任务放入队列并在 HTTP 请求主流程之外异步执行。与其同步执行繁重操作（例如，在注册时发送 1000 封邮件），不如将任务放入队列并立即向用户返回响应。后台工作进程（worker）从队列中取出任务并逐一执行。

使用队列的主要优势：

• 提升性能 — 用户无需等待繁重操作完成
• 容错性 — 如果任务执行失败，它会返回队列以便重新处理
• 可扩展性 — 可以启动任意数量的工作进程进行并行处理
• 负载缓冲 — 队列可以平滑系统峰值负载

PHP 中常用的队列工具

1. RabbitMQ

RabbitMQ 是一个强大的消息代理，支持 AMQP 协议。它非常适合复杂的路由场景，但需要安装单独的服务器。

// 通过 Composer 安装composer require php-amqplib/php-amqplib

// 将任务发送到队列<?phprequire_once 'vendor/autoload.php';

use PhpAmqpLib\\Connection\\AMQPStreamConnection;use PhpAmqpLib\\Message\\AMQPMessage;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');$channel = $connection->channel();$channel->queue_declare('email_queue', false, true, false, false);

$data = json_encode([    'to' => 'user@example.com',    'subject' => '你好！',    'body' => '测试消息']);

$msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);$channel->basic_publish($msg, '', 'email_queue');$channel->close();$connection->close();?>

2. Redis + php-resque

Redis 是一个快速的内存存储，非常适合简单的队列。php-resque 是流行的 Ruby 库 Resque 的 PHP 移植版。

// 安装composer require chrisboulton/php-resque

// 创建任务<?phpclass SendEmailJob{    public function perform()    {        $args = $this->args;        // 发送邮件的逻辑        mail($args['to'], $args['subject'], $args['body']);    }}

// 添加到队列Resque::setBackend('localhost:6379');Resque::enqueue('default', 'SendEmailJob', [    'to' => 'user@example.com',    'subject' => '你好',    'body' => '测试']);?>

3. Beanstalkd

Beanstalkd 是一个轻量级、快速且可靠的队列管理器。它易于安装和配置，是小型项目的绝佳选择。

// 安装composer require pda/pheanstalk

// 发送任务<?phprequire_once 'vendor/autoload.php';

use Pheanstalk\\Pheanstalk;

$pheanstalk = Pheanstalk::create('127.0.0.1:11300');$pheanstalk->useTube('email_tube')->put(    json_encode(['to' => 'user@example.com', 'subject' => 'Test']),    Pheanstalk::DEFAULT_PRIORITY,    Pheanstalk::DEFAULT_DELAY,    60 // TTR (time to run)