当前位置：首页 > PHP

php 实现消息队列

2026-02-14 01:10:11PHP

PHP 实现消息队列的方法

使用数据库表模拟队列

创建一个数据库表作为队列存储，通过插入和删除记录实现消息的入队和出队。

CREATE TABLE message_queue (
    id INT AUTO_INCREMENT PRIMARY KEY,
    message TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status TINYINT DEFAULT 0
);

入队操作：

$pdo->prepare("INSERT INTO message_queue (message) VALUES (?)")
    ->execute([$message]);

出队操作：

$stmt = $pdo->prepare("SELECT * FROM message_queue WHERE status = 0 ORDER BY created_at ASC LIMIT 1");
$stmt->execute();
$message = $stmt->fetch();

if ($message) {
    $pdo->prepare("UPDATE message_queue SET status = 1 WHERE id = ?")
        ->execute([$message['id']]);
    // 处理消息
}

使用 Redis 实现队列

Redis 的 List 数据结构适合实现简单的消息队列。

入队操作：

$redis->rpush('message_queue', json_encode($messageData));

出队操作：

$message = $redis->lpop('message_queue');
if ($message) {
    $data = json_decode($message, true);
    // 处理消息
}

使用专业消息队列系统

对于生产环境，推荐使用专业的消息队列系统如 RabbitMQ、Beanstalkd 或 Kafka。

RabbitMQ 示例：

$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => 5672,
    'login' => 'guest',
    'password' => 'guest'
]);
$channel = $connection->channel();
$channel->queue_declare('hello', false, false, false, false);

// 发布消息
$msg = new AMQPMessage('Hello World!');
$channel->basic_publish($msg, '', 'hello');

// 消费消息
$callback = function ($msg) {
    echo $msg->body;
};
$channel->basic_consume('hello', '', false, true, false, false, $callback);

使用 Laravel 队列系统

Laravel 提供了内置的队列系统，支持多种队列驱动。

创建任务：

php artisan make:job ProcessMessage

定义任务：

class ProcessMessage implements ShouldQueue
{
    public function __construct(public $message) {}

    public function handle()
    {
        // 处理消息
    }
}

分发任务：

ProcessMessage::dispatch($messageData);

启动队列处理器：

php artisan queue:work

使用 Supervisor 管理进程

对于长时间运行的队列处理器，建议使用 Supervisor 管理。

Supervisor 配置示例：

[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /path/to/artisan queue:work --sleep=3 --tries=3
autostart=true
autorestart=true
user=www-data
numprocs=8
redirect_stderr=true
stdout_logfile=/path/to/worker.log