本文目录导读：
在PHP中实现异步任务队列,主要有以下几种方案：
Redis + 队列（最常用）
生产者（提交任务）
// 使用 Redis 列表作为队列
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
// 添加任务到队列
$taskData = [
'type' => 'send_email',
'data' => [
'to' => 'user@example.com',
'subject' => 'Welcome',
'body' => 'Hello World'
],
'created_at' => time()
];
$redis->lpush('task_queue', json_encode($taskData));
echo "任务已提交\n";
消费者（处理任务）
// worker.php - 后台常驻进程
while (true) {
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
// 阻塞式获取任务（BRPOP 会一直等待直到有任务）
$task = $redis->brpop('task_queue', 0);
if ($task) {
$taskData = json_decode($task[1], true);
// 处理任务
try {
handleTask($taskData);
echo "[" . date('Y-m-d H:i:s') . "] 任务完成: " . $taskData['type'] . "\n";
} catch (Exception $e) {
// 记录错误日志
$redis->lpush('failed_tasks', json_encode([
'task' => $taskData,
'error' => $e->getMessage(),
'failed_at' => date('Y-m-d H:i:s')
]));
echo "任务失败: " . $e->getMessage() . "\n";
}
}
}
function handleTask($taskData) {
switch ($taskData['type']) {
case 'send_email':
// 发送邮件逻辑
sleep(2); // 模拟耗时操作
break;
case 'process_image':
// 图片处理逻辑
break;
default:
throw new Exception("未知任务类型");
}
}
使用 Beanstalkd
// 安装: composer require pda/pheanstalk
use Pheanstalk\Pheanstalk;
// 生产者
$pheanstalk = Pheanstalk::create('127.0.0.1:11300');
$tube = 'email_jobs';
$data = json_encode([
'to' => 'user@example.com',
'subject' => 'Test'
]);
$pheanstalk->useTube($tube)->put($data, 1024, 60); // 优先级, 延迟
// 消费者
while (true) {
$job = $pheanstalk->watch('email_jobs')->reserve();
$data = json_decode($job->getData(), true);
try {
// 处理任务
processEmail($data);
$pheanstalk->delete($job); // 删除完成任务
} catch (Exception $e) {
$pheanstalk->bury($job); // 任务失败放入bury
}
}
使用 RabbitMQ
// 安装: composer require php-amqplib/php-amqplib
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
// 生产者
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('task_queue', false, true, false, false);
$data = json_encode(['task' => 'send_email', 'params' => ['to' => 'a@b.com']]);
$msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
$channel->basic_publish($msg, '', 'task_queue');
$channel->close();
$connection->close();
// 消费者
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('task_queue', false, true, false, false);
$callback = function($msg) {
$data = json_decode($msg->body, true);
echo "处理任务: " . $data['task'] . "\n";
// 处理逻辑
sleep(1);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_qos(null, 1, null);
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);
while(count($channel->callbacks)) {
$channel->wait();
}
完整的异步任务管理系统
<?php
// TaskQueue.php - 完整任务队列管理系统
class TaskQueue {
private $redis;
private $queueName;
private $failedQueue;
public function __construct($host = '127.0.0.1', $port = 6379) {
$this->redis = new Redis();
$this->redis->connect($host, $port);
$this->queueName = 'task_queue';
$this->failedQueue = 'failed_queue';
}
// 添加任务
public function addTask($type, $data, $priority = 0) {
$task = [
'id' => uniqid('task_', true),
'type' => $type,
'data' => $data,
'created_at' => time(),
'priority' => $priority,
'attempts' => 0
];
// 使用有序集合存储任务，优先级作为score
$this->redis->zadd($this->queueName, $priority, json_encode($task));
return $task['id'];
}
// 获取下一个任务（带优先级）
public function getNextTask() {
$tasks = $this->redis->zrangebyscore($this->queueName, '-inf', '+inf', ['limit' => [0, 1]]);
if (empty($tasks)) {
return null;
}
$taskJson = $tasks[0];
$this->redis->zrem($this->queueName, $taskJson);
return json_decode($taskJson, true);
}
// 处理任务（带重试机制）
public function processTask($task) {
$task['attempts']++;
try {
switch ($task['type']) {
case 'email':
$this->sendEmail($task['data']);
break;
case 'image':
$this->processImage($task['data']);
break;
case 'notification':
$this->sendNotification($task['data']);
break;
default:
throw new Exception("未知任务类型");
}
// 任务成功
$this->logSuccess($task);
return true;
} catch (Exception $e) {
// 重试逻辑
if ($task['attempts'] < 3) {
$this->retryTask($task);
} else {
$this->handleFailure($task, $e->getMessage());
}
return false;
}
}
// 重试任务
private function retryTask($task) {
// 延迟重试
$delay = $task['attempts'] * 30; // 逐次增加延迟
$score = time() + $delay;
$this->redis->zadd('delayed_queue', $score, json_encode($task));
$this->logRetry($task);
}
// 处理失败任务
private function handleFailure($task, $error) {
$task['error'] = $error;
$this->redis->lpush($this->failedQueue, json_encode($task));
$this->logFailure($task, $error);
}
// 发送邮件
private function sendEmail($data) {
// 邮件发送逻辑
sleep(1);
echo "发送邮件到: " . $data['to'] . "\n";
}
// 处理图片
private function processImage($data) {
// 图片处理逻辑
sleep(2);
echo "处理图片: " . $data['path'] . "\n";
}
// 发送通知
private function sendNotification($data) {
// 通知发送逻辑
sleep(1);
echo "发送通知: " . $data['message'] . "\n";
}
// 日志记录
private function logSuccess($task) {
$logData = json_encode([
'task_id' => $task['id'],
'type' => $task['type'],
'status' => 'success',
'time' => date('Y-m-d H:i:s')
]);
$this->redis->lpush('task_logs', $logData);
}
private function logFailure($task, $error) {
$logData = json_encode([
'task_id' => $task['id'],
'type' => $task['type'],
'status' => 'failed',
'error' => $error,
'time' => date('Y-m-d H:i:s')
]);
$this->redis->lpush('task_logs', $logData);
}
private function logRetry($task) {
echo "任务 {$task['id']} 重试，第 {$task['attempts']} 次\n";
}
// 启动 Worker
public function runWorker() {
echo "Worker 开始运行...\n";
while (true) {
// 处理延迟队列
$this->processDelayedTasks();
// 获取并处理任务
$task = $this->getNextTask();
if ($task) {
echo "[" . date('Y-m-d H:i:s') . "] 处理任务: " . $task['type'] . "\n";
$this->processTask($task);
} else {
// 没有任务时休息一下
usleep(500000); // 0.5秒
}
}
}
// 处理延迟任务
private function processDelayedTasks() {
$now = time();
$tasks = $this->redis->zrangebyscore('delayed_queue', '-inf', $now);
foreach ($tasks as $taskJson) {
$this->redis->zrem('delayed_queue', $taskJson);
$task = json_decode($taskJson, true);
$this->redis->zadd($this->queueName, 0, $taskJson);
echo "任务 {$task['id']} 延迟到期，重新入队\n";
}
}
}
// 使用示例
// 提交任务
$taskQueue = new TaskQueue();
$taskId = $taskQueue->addTask('email', ['to' => 'user@example.com', 'subject' => 'Test']);
echo "任务已提交: $taskId\n";
// 启动 Worker
// $taskQueue->runWorker();
使用成熟的队列系统
Laravel Queue
// 在 Laravel 中使用
class SendEmailJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
protected $data;
public function __construct($data)
{
$this->data = $data;
}
public function handle()
{
// 处理逻辑
Mail::to($this->data['to'])->send(new WelcomeMail());
}
}
// 提交任务
SendEmailJob::dispatch($data)->onQueue('emails');
监控与管理
class TaskMonitor {
private $redis;
public function __construct() {
$this->redis = new Redis();
$this->redis->connect('127.0.0.1', 6379);
}
// 获取队列状态
public function getQueueStatus() {
return [
'pending' => $this->redis->zcard('task_queue'),
'failed' => $this->redis->llen('failed_queue'),
'delayed' => $this->redis->zcard('delayed_queue')
];
}
// 查看最近日志
public function getRecentLogs($count = 50) {
return $this->redis->lrange('task_logs', 0, $count);
}
// 重发失败任务
public function retryFailedTasks() {
while ($taskJson = $this->redis->rpop('failed_queue')) {
$task = json_decode($taskJson, true);
unset($task['error']);
$task['attempts'] = 0;
$this->redis->zadd('task_queue', 0, json_encode($task));
}
}
}
启动 Worker
# 启动单个 Worker php worker.php # 启动多个 Worker（并发） nohup php worker.php & nohup php worker.php & # 使用 supervisor 管理 # /etc/supervisor/conf.d/worker.conf [program:task_worker] command=php /path/to/worker.php numprocs=4 process_name=%(program_name)s_%(process_num)02d autostart=true autorestart=true user=www-data redirect_stderr=true
最佳实践
- 使用成熟方案：优先选择 Laravel Queues 或 RabbitMQ
- 持久化：确保任务不丢失
- 重试机制：重要任务要实现重试
- 监控告警：统计失败率并设置告警
- 并发控制：合理设置 Worker 数量
- 任务超时：设置合理的超时时间
选择哪种方案取决于你的具体需求：
- 简单场景：Redis + 队列
- 复杂需求：RabbitMQ/Beanstalkd
- 使用框架：Laravel Queues
- 需要配置管理：Beanstalkd
