php 实现消息队列

PHP 实现消息队列的方法

PHP 可以通过多种方式实现消息队列功能，以下是几种常见的方法：

使用 Redis 实现消息队列

Redis 的 List 数据结构非常适合实现简单的消息队列。可以使用 LPUSH 和 BRPOP 命令实现生产者-消费者模式。

生产者代码示例：

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->lPush('message_queue', 'Hello, World!');

消费者代码示例：

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
while (true) {
    $message = $redis->brPop('message_queue', 0);
    echo "Received: " . $message[1] . "\n";
}

使用 RabbitMQ 实现消息队列

RabbitMQ 是一个功能强大的消息代理，PHP 可以通过 AMQP 扩展与之交互。

安装 AMQP 扩展：

pecl install amqp

生产者代码示例：

$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => 5672,
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect();
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName('messages');
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->declareExchange();
$exchange->publish('Hello, World!', 'routing_key');

消费者代码示例：

$connection = new AMQPConnection([...]);
$connection->connect();
$channel = new AMQPChannel($connection);
$queue = new AMQPQueue($channel);
$queue->setName('message_queue');
$queue->declareQueue();
$queue->bind('messages', 'routing_key');
$queue->consume(function ($message) {
    echo "Received: " . $message->getBody() . "\n";
    return true;
});

使用数据库实现消息队列

对于简单的场景，可以使用数据库表作为消息队列。

创建消息表：

CREATE TABLE message_queue (
    id INT AUTO_INCREMENT PRIMARY KEY,
    message TEXT NOT NULL,
    status ENUM('pending', 'processed') DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

生产者代码示例：

$pdo = new PDO('mysql:host=localhost;dbname=test', 'username', 'password');
$stmt = $pdo->prepare("INSERT INTO message_queue (message) VALUES (?)");
$stmt->execute(['Hello, World!']);

消费者代码示例：

$pdo = new PDO('mysql:host=localhost;dbname=test', 'username', 'password');
$stmt = $pdo->prepare("SELECT * FROM message_queue WHERE status = 'pending' LIMIT 1 FOR UPDATE");
$pdo->beginTransaction();
$stmt->execute();
$message = $stmt->fetch();
if ($message) {
    echo "Processing: " . $message['message'] . "\n";
    $update = $pdo->prepare("UPDATE message_queue SET status = 'processed' WHERE id = ?");
    $update->execute([$message['id']]);
    $pdo->commit();
}

使用 Beanstalkd 实现消息队列

Beanstalkd 是一个轻量级的工作队列系统。

安装 Beanstalkd：

sudo apt-get install beanstalkd

生产者代码示例：

$pheanstalk = new Pheanstalk\Pheanstalk('127.0.0.1');
$pheanstalk->useTube('test')->put('Hello, World!');

消费者代码示例：

$pheanstalk = new Pheanstalk\Pheanstalk('127.0.0.1');
$job = $pheanstalk->watch('test')->ignore('default')->reserve();
echo "Processing: " . $job->getData() . "\n";
$pheanstalk->delete($job);

使用 Laravel 队列系统

如果使用 Laravel 框架，可以利用其内置的队列系统。

配置队列驱动（如 Redis）：

QUEUE_CONNECTION=redis

创建任务：

php artisan make:job ProcessMessage

定义任务逻辑：

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

    public function handle()
    {
        echo "Processing: " . $this->message . "\n";
    }
}

分发任务：

ProcessMessage::dispatch('Hello, World!');

启动队列处理器：

php artisan queue:work

选择合适的方法

• 对于简单应用，Redis 或数据库方案足够
• 对于需要持久化和高级特性的场景，RabbitMQ 或 Beanstalkd 更合适
• 在 Laravel 项目中，使用内置队列系统最为方便

每种方法都有其优缺点，应根据项目需求、性能要求和运维复杂度进行选择。