RabbitMQ 分布式系统实战：从安装部署到 C++ 客户端调用

技术对比

安装与配置

服务端安装

1. 安装 RabbitMQ

sudo apt install rabbitmq-server

2. 启动 & 检查状态

# 启动服务
sudo systemctl start rabbitmq-server
# 查看状态（确保 active (running)）
sudo systemctl status rabbitmq-server

如果看到服务运行状态正常，说明安装成功。

3. 创建管理员用户

默认的 guest 用户权限较低且仅限本地登录，建议创建专用管理员。

# 添加用户（示例：用户名 root，密码 123456）
sudo rabbitmqctl add_user root 123456
# 设为管理员
sudo rabbitmqctl set_user_tags root administrator
# 赋予权限（对所有虚拟主机）
sudo rabbitmqctl set_permissions -p / root ".*" ".*" ".*"

4. 开启 Web 管理界面

默认端口为 15672。

sudo rabbitmq-plugins enable rabbitmq_management

访问地址：http://你的服务器 IP:15672，使用刚才创建的用户名和密码登录。

注意：如果无法访问，可能需要修改配置文件开放端口。编辑 /etc/rabbitmq/rabbitmq.conf，添加监听配置，然后重启服务。

常见问题：登录 Web UI 时报错通常不是账号密码问题，而是权限未正确设置。重新执行上述用户创建命令并重启服务即可解决。

客户端及 C++ 库安装

1. 安装官方 C 客户端库

提供基础的 RabbitMQ C 接口。

sudo apt-get install librabbitmq-dev

2. 安装 C++ 客户端库（AMQP-CPP）

这是基于 C 接口的 C++ 封装库，更适合 C++ 项目。

(1) 安装依赖库 libev

sudo apt install libev-dev

(2) 下载源码

git clone https://github.com/CopernicaMarketingSoftware/AMQP-CPP.git
cd AMQP-CPP/

(3) 编译并安装

make && make install

注意：如果编译时出现 SSL 版本错误，尝试修复系统依赖或重新执行 make。若遇到冲突的 SSL 库，可强制卸载相关包后再试。

端口说明：

• Web 控制台默认 15672（可根据配置修改）。
• 服务端通信默认 5672。
• 集群模式端口通常为 25672。

AMQP-CPP 库详解

库简介

AMQP-CPP 是一个高性能的 C++ 库，专门用于与 RabbitMQ 通信。它完全异步，无阻塞调用，适合对性能要求高的场景，需要 C++17 或更高版本编译。

两种工作模式

模式 1：TCP 模式（自己管网络）

• 核心操作：需自己实现一个类，继承自 AMQP::TcpHandler。
• 关键步骤：重写 monitor 函数，将网络套接字交给自己的事件循环（如 select/epoll），当 fd 可读/可写时调用 connection->process(fd, flags)。
• 适用场景：想完全自定义网络层，使用熟悉的网络库。

模式 2：扩展模式（用现成库省事）

• 推荐方式：直接使用 AMQP-CPP 提供的 AMQP::LibEvHandler 类。
• 优势：不需要自己写 monitor 函数，库会自动处理 fd 监控和消息处理。
• 其他支持：还支持 libevent、libuv、asio 等异步通信组件。
• 适用场景：快速集成，不想手动写网络 IO 代码。

后续示例将采用扩展模式（libev）来实现。

常用接口

• Channel: 信道类，用于声明交换机、队列、绑定、发布消息、订阅消息等。
  • declareExchange(): 声明交换机
  • declareQueue(): 声明队列
  • bindQueue(): 绑定交换机与队列
  • publish(): 发布消息
  • consume(): 订阅队列消息
  • ack(): 消息确认应答
• Message: 消息类，获取消息正文及大小。
• libev: 事件循环结构体与接口，用于底层 I/O 监控。

实战演示

测试思路

分别实现发布客户端与订阅客户端，底层关联事件监控，建立连接、信道，声明交换机、队列、绑定，最后进行消息发布与订阅。

效果验证

启动发布客户端后，可以看到连接与信道建立成功。在 Web 控制台中也能观察到对应的交换机、队列以及发送进来的消息。启动订阅客户端后，能够收到对应消息，服务端消息计数也会更新。

注意：原生 RabbitMQ 无论是先启动订阅还是发布端，最后订阅端都能收到发进去的消息。这与某些仿制实现不同，后者可能要求必须先启动订阅端才能接收推送。

代码示例

以下是发布 (publish.cc) 与订阅 (consume.cc) 的核心逻辑示意，具体实现请参考代码注释。

// 伪代码示意，实际请参照完整工程
#include <amqpcpp.h>

int main() {
    // 初始化 loop
    struct ev_loop *loop = EV_DEFAULT;
    
    // 建立连接
    AMQP::Connection connection(loop, "amqp://root:123456@localhost:5672/");
    
    // 声明通道
    AMQP::Channel channel(&connection);
    
    // 声明交换机与队列
    channel.declareExchange("my_exchange", AMQP::exchange_direct).then([](const AMQP::DeclareExchangeResult &result){
        // 绑定队列
        channel.bindQueue("my_queue", "my_exchange", "routing_key");
    });
    
    // 运行事件循环
    ev_run(loop, 0);
    return 0;
}

API 二次封装

为了简化开发，我们可以对 AMQP-CPP 进行二次封装。

封装思想

1. 将事件监控结构 loop、通信连接 connection 以及信道 channel 封装起来。
2. 提供创建交换机、队列并进行绑定的便捷接口。
3. 提供向指定交换机发布消息的接口。
4. 提供订阅指定队列消息的接口。

注意事项

• 声明完交换机和队列后才回去调用回调处理，可能存在时序差异，需注意发布顺序。
• 主线程阻塞等待异步线程完成处理后，通过异步事件通知来 break 循环进行回收。

总结

RabbitMQ 作为强大的消息队列中间件，通过灵活的交换机和丰富路由规则实现异步通信、解耦等功能。Linux 下安装便捷，AMQP-CPP 库提供高效的 C++ 交互方式，配合客户端 API 封装能显著提升开发效率，适用于多种分布式场景。