PHP异步任务队列怎么做

wen PHP项目 2026-08-11 140

本文目录导读：

PHP异步任务队列怎么做

Redis + 队列（最常用）
使用 Beanstalkd
使用 RabbitMQ
完整的异步任务管理系统
使用成熟的队列系统
监控与管理
启动 Worker
最佳实践

在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