PHP实现消息队列

封面图 • 2025-09-18 • PHP

在 PHP 中实现消息队列（Message Queue）有多种方式，可以根据实际需求选择不同的实现方式。以下是一些常见的 PHP 消息队列实现方法：

---

## ✅ 一、使用 PHP 自带的 `pcntl` 和 `shmop` 实现简单的消息队列

这是最基础的方式，适用于小型项目或学习用途。

### 示例代码：

```php
<?php

// 生产者：将消息放入队列
function producer($message) {
    $key = ftok(__FILE__, 'a'); // 获取共享内存键
    $shm_id = shmop_open($key, "c", 0644, 1024); // 创建或打开共享内存段

if (!$shm_id) {
        die("无法创建共享内存");
    }

$size = shmop_write($shm_id, $message, 0);
    if ($size === false) {
        die("写入失败");
    }

echo "消息已放入队列: $message\n";
    shmop_close($shm_id);
}

// 消费者：从队列中取出消息
function consumer() {
    $key = ftok(__FILE__, 'a');
    $shm_id = shmop_open($key, "a", 0644, 1024);

if (!$shm_id) {
        die("无法打开共享内存");
    }

$data = shmop_read($shm_id, 0, 1024);
    if ($data === false) {
        die("读取失败");
    }

echo "消息已取出: " . trim($data) . "\n";
    shmop_close($shm_id);
}

// 测试
producer("Hello, Queue!");
consumer();
```

> ⚠️ 注意：这种方式仅适用于单机环境，且不支持持久化、并发控制等高级功能。

---

## ✅ 二、使用 Redis 实现消息队列（推荐）

Redis 提供了 `LPUSH` 和 `RPOP` 等命令来实现消息队列，适合大多数中小型应用。

### 安装 Redis

确保你已经安装并启动了 Redis 服务。

### 示例代码：

#### 生产者（发送消息）：

```php
<?php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$queueName = 'my_queue';
$message = 'Hello from PHP queue';

$redis->rPush($queueName, $message);
echo "消息已发送: $message\n";
```

#### 消费者（接收消息）：

```php
<?php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$queueName = 'my_queue';

while (true) {
    $message = $redis->lPop($queueName);
    if ($message) {
        echo "收到消息: $message\n";
    } else {
        echo "队列为空，等待中...\n";
        sleep(1); // 等待一段时间再尝试
    }
}
```

> ✅ 优点：支持持久化、高并发、跨进程/跨服务器通信。

---

## ✅ 三、使用 RabbitMQ 实现消息队列（推荐用于复杂系统）

RabbitMQ 是一个强大的消息中间件，支持多种协议和高级特性。

### 安装 RabbitMQ

参考官方文档安装：https://www.rabbitmq.com/download.html

### 示例代码（使用 PHP 的 `php-amqplib` 库）：

#### 安装依赖：

```bash
composer require php-amqplib/php-amqplib
```

#### 生产者：

```php
<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('my_queue', false, false, false, false);

$msg = new AMQPMessage('Hello from PHP queue');
$channel->basic_publish($msg, '', 'my_queue');

echo "消息已发送\n";

$channel->close();
$connection->close();
```

#### 消费者：

```php
<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('my_queue', false, false, false, false);

echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

$callback = function ($msg) {
    echo " [x] Received ", $msg->body, "\n";
};

$channel->basic_consume('my_queue', '', false, true, false, false, $callback);

while ($channel->is_open()) {
    $channel->wait();
}
```

> ✅ 优点：支持可靠消息传递、延迟队列、优先级队列等高级功能。

---

## ✅ 四、使用 Beanstalkd（轻量级消息队列）

Beanstalkd 是一个轻量级的消息队列系统，适合需要高性能和简单部署的场景。

### 安装 Beanstalkd

```bash
sudo apt-get install beanstalkd
```

### 示例代码（使用 `pda/pheanstalk` 库）：

#### 安装依赖：

```bash
composer require pda/pheanstalk
```

#### 生产者：

```php
<?php
require 'vendor/autoload.php';

use Pheanstalk\Pheanstalk;

$pheanstalk = Pheanstalk::create('127.0.0.1');

$pheanstalk->put("Hello from PHP queue");
echo "消息已发送\n";
```

#### 消费者：

```php
<?php
require 'vendor/autoload.php';

use Pheanstalk\Pheanstalk;

$pheanstalk = Pheanstalk::create('127.0.0.1');

while (true) {
    $job = $pheanstalk->reserve();
    echo "收到消息: " . $job->getData() . "\n";
    $pheanstalk->delete($job);
}
```

---

## ✅ 五、使用 Laravel 的队列系统（如果你用 Laravel）

Laravel 内置了强大的队列系统，支持多种驱动（如 Redis、Beanstalkd、Database、Sqs 等）。

### 示例代码（使用 Redis 驱动）：

#### 生成任务类：

```bash
php artisan make:job ProcessMessage
```

#### 编写任务逻辑：

```php
<?php

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

class ProcessMessage implements ShouldQueue
{
    use InteractsWithQueue, Queueable, SerializesModels;

protected $message;

public function __construct($message)
    {
        $this->message = $message;
    }

public function handle()
    {
        echo "处理消息: " . $this->message . "\n";
    }
}
```

#### 发送任务：

```php
ProcessMessage::dispatch('Hello from Laravel queue');
```

#### 启动消费者：

```bash
php artisan queue:work
```

---

## ✅ 总结

| 方法 | 是否支持持久化 | 是否支持分布式 | 适用场景 |
|------|----------------|----------------|----------|
| `shmop` + `pcntl` | ❌ | ❌ | 小型/学习项目 |
| Redis | ✅ | ✅ | 中小型项目 |
| RabbitMQ | ✅ | ✅ | 复杂系统、企业级 |
| Beanstalkd | ✅ | ✅ | 轻量级、高性能 |
| Laravel Queue | ✅ | ✅ | Laravel 项目 |

---

如果你告诉我你的具体需求（比如是否需要分布式、是否要持久化、是否使用 Laravel 等），我可以帮你选择最合适的一种方案。