当前位置：首页 > PHP

php队列实现方式

2026-02-15 10:01:06PHP

PHP 队列的实现方式

PHP 队列的实现可以通过多种方式完成，包括使用数据库、Redis、消息队列服务（如 RabbitMQ、Kafka）或专门的队列管理工具（如 Laravel 的队列系统）。以下是几种常见的实现方式：

使用数据库实现队列

数据库是最简单的队列实现方式之一，适合小型应用或开发初期。可以创建一个表来存储队列任务。

CREATE TABLE queue_jobs (
    id INT AUTO_INCREMENT PRIMARY KEY,
    job_name VARCHAR(255) NOT NULL,
    payload TEXT,
    attempts INT DEFAULT 0,
    reserved_at TIMESTAMP NULL,
    available_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

PHP 代码示例：

// 入队
$pdo->prepare("INSERT INTO queue_jobs (job_name, payload) VALUES (?, ?)")
    ->execute(['send_email', json_encode(['to' => 'user@example.com'])]);

// 出队
$pdo->beginTransaction();
$job = $pdo->query("SELECT * FROM queue_jobs WHERE reserved_at IS NULL ORDER BY created_at LIMIT 1 FOR UPDATE")->fetch();
if ($job) {
    $pdo->prepare("UPDATE queue_jobs SET reserved_at = NOW() WHERE id = ?")->execute([$job['id']]);
    $pdo->commit();
    // 处理任务
    processJob($job);
} else {
    $pdo->rollBack();
}

使用 Redis 实现队列

Redis 提供了高效的内存队列支持，适合高吞吐量的场景。可以使用 LPUSH 和 RPOP 命令实现队列。

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// 入队
$redis->lPush('queue:emails', json_encode(['to' => 'user@example.com']));

// 出队
$job = $redis->rPop('queue:emails');
if ($job) {
    processJob(json_decode($job, true));
}

使用 Laravel 队列系统

Laravel 提供了强大的队列系统，支持数据库、Redis、Amazon SQS 等多种驱动。

定义任务：

php artisan make:job SendEmail

在 SendEmail 类中实现 handle 方法：

public function handle()
{
    Mail::to($this->email)->send(new WelcomeEmail());
}

分发任务：

SendEmail::dispatch($email);

启动队列处理器：

php artisan queue:work

使用 RabbitMQ 实现队列

RabbitMQ 是一个功能强大的消息队列服务，适合分布式系统。

安装 PHP 的 AMQP 扩展后，示例代码：

$connection = new AMQPConnection(['host' => 'localhost']);
$connection->connect();
$channel = new AMQPChannel($connection);
$queue = new AMQPQueue($channel);
$queue->setName('emails');
$queue->declareQueue();

// 入队
$queue->publish(json_encode(['to' => 'user@example.com']));

// 出队
$message = $queue->get();
if ($message) {
    processJob(json_decode($message->getBody(), true));
    $queue->ack($message->getDeliveryTag());
}

使用 Kafka 实现队列

Kafka 适合高吞吐量的消息队列场景。

安装 php-rdkafka 扩展后，示例代码：

$conf = new RdKafka\Conf();
$conf->set('group.id', 'email_consumers');
$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe(['emails']);

while (true) {
    $message = $consumer->consume(120*1000);
    if ($message->err) {
        continue;
    }
    processJob(json_decode($message->payload, true));
}