本文目录导读：
我来详细介绍PHP项目中使用RabbitMQ的方法。
安装RabbitMQ PHP客户端
使用Composer安装
composer require php-amqplib/php-amqplib
安装RabbitMQ服务（如果未安装）
# Ubuntu/Debian sudo apt-get install rabbitmq-server # CentOS/RHEL sudo yum install rabbitmq-server # 启动服务 sudo systemctl start rabbitmq-server sudo systemctl enable rabbitmq-server
基本使用示例
生产者（发送消息）
<?php
require_once 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class RabbitMQProducer {
private $connection;
private $channel;
private $exchange;
private $queue;
public function __construct($config = []) {
$defaultConfig = [
'host' => 'localhost',
'port' => 5672,
'user' => 'guest',
'password' => 'guest',
'vhost' => '/'
];
$config = array_merge($defaultConfig, $config);
try {
$this->connection = new AMQPStreamConnection(
$config['host'],
$config['port'],
$config['user'],
$config['password'],
$config['vhost']
);
$this->channel = $this->connection->channel();
} catch (\Exception $e) {
throw new \Exception("RabbitMQ连接失败: " . $e->getMessage());
}
}
public function sendMessage($queue, $message, $exchange = '') {
// 声明队列
$this->channel->queue_declare(
$queue, // 队列名
false, // passive
true, // durable（持久化）
false, // exclusive
false // auto_delete
);
// 创建消息
$msg = new AMQPMessage(
json_encode($message),
[
'content_type' => 'application/json',
'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
]
);
// 发送消息
$this->channel->basic_publish(
$msg,
$exchange,
$queue
);
echo " [x] 发送消息: " . json_encode($message) . "\n";
}
public function __destruct() {
$this->channel->close();
$this->connection->close();
}
}
// 使用示例
$producer = new RabbitMQProducer();
$producer->sendMessage('task_queue', [
'id' => 1,
'type' => 'email',
'data' => [
'to' => 'user@example.com',
'subject' => 'Test Subject',
'body' => 'Hello World'
]
]);
消费者（接收消息）
<?php
require_once 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
class RabbitMQConsumer {
private $connection;
private $channel;
public function __construct($config = []) {
$defaultConfig = [
'host' => 'localhost',
'port' => 5672,
'user' => 'guest',
'password' => 'guest',
'vhost' => '/'
];
$config = array_merge($defaultConfig, $config);
try {
$this->connection = new AMQPStreamConnection(
$config['host'],
$config['port'],
$config['user'],
$config['password'],
$config['vhost']
);
$this->channel = $this->connection->channel();
} catch (\Exception $e) {
throw new \Exception("RabbitMQ连接失败: " . $e->getMessage());
}
}
public function consume($queue, $callback) {
// 声明队列
$this->channel->queue_declare(
$queue,
false,
true, // durable
false,
false
);
// 设置每次只取一条消息
$this->channel->basic_qos(
null, // prefetch_size
1, // prefetch_count
null // a_global
);
echo " [*] 等待消息，按Ctrl+C退出\n";
// 消费消息
$this->channel->basic_consume(
$queue,
'', // consumer_tag
false, // no_local
false, // no_ack
false, // exclusive
false, // nowait
function($msg) use ($callback) {
echo " [x] 收到消息: " . $msg->body . "\n";
// 处理消息
$data = json_decode($msg->body, true);
$result = $callback($data);
// 确认消息已处理
$msg->delivery_info['channel']->basic_ack(
$msg->delivery_info['delivery_tag']
);
}
);
// 循环等待消息
while ($this->channel->is_consuming()) {
$this->channel->wait();
}
}
public function __destruct() {
$this->channel->close();
$this->connection->close();
}
}
// 使用示例
$consumer = new RabbitMQConsumer();
$consumer->consume('task_queue', function($data) {
// 处理业务逻辑
echo "处理任务: " . $data['type'] . "\n";
// 模拟耗时任务
sleep(2);
echo "任务完成: " . $data['id'] . "\n";
return true;
});
高级用法
工作队列（任务分发）
// 生产者 - 分发任务
class TaskProducer {
public function sendTasks() {
$producer = new RabbitMQProducer();
for ($i = 0; $i < 10; $i++) {
$task = [
'id' => $i,
'type' => 'task',
'data' => "任务内容 $i",
'timestamp' => time()
];
$producer->sendMessage('task_queue', $task);
}
}
}
// 消费者 - 处理任务（可启动多个实例）
class TaskWorker {
public function start() {
$consumer = new RabbitMQConsumer();
$consumer->consume('task_queue', function($task) {
// 处理任务
echo "Worker " . getmypid() . " 处理任务: " . $task['id'] . "\n";
// 模拟处理时间
sleep(rand(1, 3));
return true;
});
}
}
发布/订阅模式
// 发布者
class Publisher {
public function publish($exchange, $message) {
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
// 声明交换机
$channel->exchange_declare(
$exchange,
'fanout', // 类型：fanout, direct, topic, headers
false,
false,
false
);
$msg = new AMQPMessage(json_encode($message));
$channel->basic_publish($msg, $exchange);
echo " [x] 发布消息到交换机 $exchange\n";
$channel->close();
$connection->close();
}
}
// 订阅者
class Subscriber {
public function subscribe($exchange, $queueName = '') {
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
// 声明交换机
$channel->exchange_declare(
$exchange,
'fanout',
false,
false,
false
);
// 创建临时队列
list($queueName, ,) = $channel->queue_declare(
$queueName,
false,
false,
true, // exclusive
false
);
// 绑定队列到交换机
$channel->queue_bind($queueName, $exchange);
echo " [*] 等待日志消息，按Ctrl+C退出\n";
$channel->basic_consume($queueName, '', false, true, false, false, function($msg) {
echo " [x] 收到: " . $msg->body . "\n";
});
while ($channel->is_consuming()) {
$channel->wait();
}
}
}
路由模式（Direct Exchange）
// 生产者
class DirectProducer {
public function send($routingKey, $message) {
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->exchange_declare('direct_logs', 'direct', false, false, false);
$msg = new AMQPMessage(json_encode($message));
$channel->basic_publish($msg, 'direct_logs', $routingKey);
echo " [x] 发送 [$routingKey]: " . json_encode($message) . "\n";
$channel->close();
$connection->close();
}
}
// 消费者
class DirectConsumer {
public function consume($routingKeys) {
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->exchange_declare('direct_logs', 'direct', false, false, false);
list($queueName, ,) = $channel->queue_declare("", false, false, true, false);
// 绑定多个路由键
foreach ($routingKeys as $routingKey) {
$channel->queue_bind($queueName, 'direct_logs', $routingKey);
}
echo " [*] 等待消息，按Ctrl+C退出\n";
$channel->basic_consume($queueName, '', false, true, false, false, function($msg) {
echo " [x] " . $msg->delivery_info['routing_key'] . ":" . $msg->body . "\n";
});
while ($channel->is_consuming()) {
$channel->wait();
}
}
}
配置文件和最佳实践
配置文件（config/rabbitmq.php）
<?php
return [
'host' => env('RABBITMQ_HOST', 'localhost'),
'port' => env('RABBITMQ_PORT', 5672),
'user' => env('RABBITMQ_USER', 'guest'),
'password' => env('RABBITMQ_PASSWORD', 'guest'),
'vhost' => env('RABBITMQ_VHOST', '/'),
'options' => [
'heartbeat' => 60,
'read_write_timeout' => 120,
],
'queues' => [
'task_queue' => [
'name' => 'task_queue',
'durable' => true,
],
'email_queue' => [
'name' => 'email_queue',
'durable' => true,
],
'notification_queue' => [
'name' => 'notification_queue',
'durable' => true,
],
],
'exchanges' => [
'logs' => [
'name' => 'logs',
'type' => 'fanout',
],
'direct_logs' => [
'name' => 'direct_logs',
'type' => 'direct',
],
],
];
连接池管理
<?php
class RabbitMQPool {
private static $instance = null;
private $connections = [];
private $config;
private function __construct($config) {
$this->config = $config;
}
public static function getInstance($config = []) {
if (self::$instance === null) {
self::$instance = new self($config);
}
return self::$instance;
}
public function getConnection() {
// 简单的连接复用
if (!empty($this->connections)) {
return array_pop($this->connections);
}
return new AMQPStreamConnection(
$this->config['host'],
$this->config['port'],
$this->config['user'],
$this->config['password']
);
}
public function releaseConnection($connection) {
$this->connections[] = $connection;
}
}
常见问题处理
连接失败处理
try {
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
} catch (\Exception $e) {
// 记录日志
error_log("RabbitMQ连接失败: " . $e->getMessage());
// 重试逻辑
$retryCount = 0;
$maxRetries = 3;
while ($retryCount < $maxRetries) {
try {
sleep(2);
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
break;
} catch (\Exception $e) {
$retryCount++;
if ($retryCount >= $maxRetries) {
throw new \Exception("RabbitMQ连接失败，已重试{$maxRetries}次");
}
}
}
}
消息确认和重试
// 消费者中使用手动确认
$channel->basic_consume($queue, '', false, false, false, false, function($msg) {
try {
// 处理消息
processMessage(json_decode($msg->body, true));
// 确认消息
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
} catch (\Exception $e) {
// 处理失败，拒绝消息并可选择重新入队
$msg->delivery_info['channel']->basic_nack(
$msg->delivery_info['delivery_tag'],
false,
true // requeue
);
}
});
这些示例涵盖了RabbitMQ在PHP项目中的基本使用和高级用法,根据你的具体需求选择合适的模式进行实现。
