详解PHP多进程消费队列

在实际生产环境中，我们经常需要处理异步任务（如发送短信、邮件、消息推送等）。常见的解决方案是使用队列实现异步任务的处理。本文将详解如何使用PHP多进程消费队列，实现高效、可靠的异步任务处理。

一、队列的基本概念

• 队列：一种先进先出（FIFO）的数据结构。
• 生产者：向队列中放入任务的部分。
• 消费者：从队列中获取任务并进行处理的部分。

二、多进程消费队列

1. 采用Redis做消息中间件，实现队列的生产、消费。使用可靠性更高的rpush、blpop进行队列任务的压入和弹出。

// 引入redis类库
use Predis\Client as Redis;

// 连接Redis
$redis = new Redis([
    'scheme' => 'tcp',
    'host'   => 'redis',
    'port'   => 6379,
]);

// 压入任务
$redis->rpush('queue_name', 'task1');
$redis->rpush('queue_name', 'task2');

// 弹出任务
$task = $redis->blpop('queue_name', 0)[1];

1. 多进程处理队列任务，提高消费速度。使用pcntl_fork()实现多进程管理。

// 处理队列任务
function handleQueueTask($redis)
{
    // 从队列中获取任务
    $task = $redis->blpop('queue_name', 0)[1];

    // 处理任务
    // ...
}

// 创建进程
for ($i = 0; $i < $processNum; $i++) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die('fork failed');
    } elseif ($pid == 0) {
        // 子进程处理任务
        handleQueueTask($redis);
        exit(0);
    }
}

// 等待子进程退出
for ($i = 0; $i < $processNum; $i++) {
    pcntl_wait($status);
}

三、完整示例

下面是一个完整的示例代码，实现了异步发送邮件功能。通过将邮件任务放入Redis队列，并使用多进程处理邮件任务，实现高效的邮件发送。

1. 生产者代码

// 连接Redis
$redis = new Redis([
    'scheme' => 'tcp',
    'host'   => 'redis',
    'port'   => 6379,
]);

// 添加邮件任务到队列中
$mailData = [
    'to' => '[e-mail]',
    'subject' => 'Test email',
    'body' => 'This is a test email from queue',
];
$redis->rpush('mail_queue', json_encode($mailData));

2. 消费者代码

// 处理邮件任务
function handleMailTask($redis, $mailData)
{
    // 发送邮件
    $mail = new PHPMailer\PHPMailer\PHPMailer(true);
    $mail->Host = 'mail.example.com';
    $mail->SMTPAuth = true;
    $mail->Username = '[e-mail]';
    $mail->Password = 'password';
    $mail->SMTPSecure = 'tls';
    $mail->Port = 587;
    $mail->From = '[e-mail]';
    $mail->FromName = 'Example';
    $mail->addAddress($mailData['to']);
    $mail->isHTML(true);
    $mail->Subject = $mailData['subject'];
    $mail->Body = $mailData['body'];
    $mail->send();
}

// 创建进程处理邮件任务
$processNum = 5;
for ($i = 0; $i < $processNum; $i++) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die('fork failed');
    } elseif ($pid == 0) {
        $redis = new Redis([
            'scheme' => 'tcp',
            'host'   => 'redis',
            'port'   => 6379,
        ]);
        while (true) {
            // 从邮件队列中获取任务
            $mailData = json_decode($redis->blpop('mail_queue', 0)[1], true);
            // 处理邮件任务
            handleMailTask($redis, $mailData);
        }
        exit(0);
    }
}

// 等待子进程退出
for ($i = 0; $i < $processNum; $i++) {
    pcntl_wait($status);
}

四、总结

本文介绍了如何使用PHP多进程消费队列，实现高效、可靠的异步任务处理。我们将队列放入Redis中，并使用子进程处理队列任务，大大提高了任务的处理速度。希望本文对您有所帮助。