当前位置：首页 > PHP

php队列的实现

2026-02-13 13:15:33PHP

PHP 队列的实现方法

PHP 队列可以通过多种方式实现，包括数据库驱动、文件驱动、内存驱动以及第三方服务。以下是几种常见的实现方法：

使用数据库实现队列

数据库是最简单的队列实现方式之一，适合小型应用或开发环境。

// 创建队列表
CREATE TABLE queue (
    id INT AUTO_INCREMENT PRIMARY KEY,
    payload TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    processed_at TIMESTAMP NULL,
    status ENUM('pending', 'processing', 'completed') DEFAULT 'pending'
);

// 入队操作
$payload = json_encode(['task' => 'send_email', 'data' => ['to' => 'user@example.com']]);
$stmt = $pdo->prepare("INSERT INTO queue (payload, status) VALUES (?, 'pending')");
$stmt->execute([$payload]);

// 出队操作
$pdo->beginTransaction();
$stmt = $pdo->prepare("SELECT * FROM queue WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1 FOR UPDATE");
$stmt->execute();
$job = $stmt->fetch(PDO::FETCH_ASSOC);

if ($job) {
    $updateStmt = $pdo->prepare("UPDATE queue SET status = 'processing', processed_at = NOW() WHERE id = ?");
    $updateStmt->execute([$job['id']]);
    $pdo->commit();
    // 处理任务
    $payload = json_decode($job['payload'], true);
    // ... 执行任务逻辑
    $completeStmt = $pdo->prepare("UPDATE queue SET status = 'completed' WHERE id = ?");
    $completeStmt->execute([$job['id']]);
} else {
    $pdo->rollBack();
}

使用 Redis 实现队列

Redis 提供了高性能的内存队列支持，适合高吞吐量场景。

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

// 入队操作
$redis->lPush('queue', json_encode(['task' => 'process_image', 'data' => ['image_id' => 123]]));

// 出队操作
while (true) {
    $job = $redis->rPop('queue');
    if ($job) {
        $data = json_decode($job, true);
        // ... 处理任务逻辑
    } else {
        sleep(1); // 队列为空时等待
    }
}

使用专业队列系统

对于生产环境，建议使用专业的队列系统如 RabbitMQ 或 Beanstalkd。

// 使用 Beanstalkd
$queue = new Pheanstalk\Pheanstalk('127.0.0.1');

// 入队
$queue->useTube('default')
    ->put(json_encode(['task' => 'generate_report']));

// 出队
$job = $queue->watch('default')
    ->ignore('default')
    ->reserve();

$data = json_decode($job->getData(), true);
// ... 处理任务
$queue->delete($job);

使用 Laravel 队列系统

Laravel 提供了开箱即用的队列系统，支持多种后端驱动。

// 创建任务类
php artisan make:job ProcessPodcast

// 任务类示例
class ProcessPodcast implements ShouldQueue
{
    public function __construct(public Podcast $podcast) {}

    public function handle()
    {
        // 处理播客...
    }
}

// 分发任务
ProcessPodcast::dispatch($podcast);

// 启动队列 worker
php artisan queue:work

使用 Symfony Messenger

Symfony 的 Messenger 组件提供了灵活的队列功能。

// 配置 messenger.yaml
framework:
    messenger:
        transports:
            async: '%env(MESSENGER_TRANSPORT_DSN)%'
        routing:
            'App\Message\MyMessage': async

// 创建消息
class MyMessage
{
    public function __construct(private string $content) {}

    public function getContent(): string
    {
        return $this->content;
    }
}

// 发送消息
$bus->dispatch(new MyMessage('Hello!'));

// 处理消息
class MyMessageHandler
{
    public function __invoke(MyMessage $message)
    {
        // ... 处理消息
    }
}