PHP项目RabbitMQ如何PHP端使用

wen PHP项目 2026-07-01 115

本文目录导读：

安装RabbitMQ PHP客户端
基本使用示例
高级用法
配置文件和最佳实践
常见问题处理

我来详细介绍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项目中的基本使用和高级用法,根据你的具体需求选择合适的模式进行实现。

本文地址： https://www.bmjup.com/post/13615.html