在即时通讯项目中搭建 RabbitMQ 消息中间件及使用 AMQP-CPP C++ 库的过程。内容包括 RabbitMQ 的功能优势、安装配置步骤、AMQP-CPP 的安装方法以及三个具体的代码示例（基础收发、分离发送接收、交换机绑定）。此外，文章还展示了如何对 AMQP-CPP 进行二次封装，实现连接管理、组件声明、消息发布与消费的统一接口，并详细解释了路由键匹配机制及直连交换机的测试验证。通过 libev 事件循环与多线程结合，确保了高性能异步通信的实现。
RabbitMQ
RabbitMQ 是一个消息中间件，你可以把它理解成一个专门负责接收、存储和转发消息的程序。它让不同的软件系统或者同一个系统的不同模块之间可以相互通信，但不需要直接连接对方。
它的工作方式很简单：
这样做的好处是：
所以，RabbitMQ 常被用在需要可靠通信的场景，比如电商订单处理、日志收集、任务调度等。它就像是系统之间的'信使'，帮忙传递信息，让整个系统更灵活、更稳定。
AMQP-CPP
AMQP-CPP 是一个用于与 RabbitMQ 消息中间件通信的 C++ 库。RabbitMQ 是一个广泛使用的开源消息代理，它实现了 AMQP（高级消息队列协议）。
简单来说，RabbitMQ 负责在不同应用之间传递消息，而 AMQP-CPP 则帮助 C++ 程序与 RabbitMQ 进行交互。
AMQP-CPP 的核心设计理念
异步与高性能
对 C++ 版本的要求
RabbitMQ 安装
安装 RabbitMQ
sudo apt install -y rabbitmq-server
# 启动 RabbitMQ 服务
sudo systemctl start rabbitmq-server.service
# 查看 RabbitMQ 服务状态，确认是否正常运行
sudo systemctl status rabbitmq-server.service
# 设置为开机自启动
sudo systemctl enable rabbitmq-server
安装完成后默认有一个 guest 用户，但权限不足，无法用于远程登录和消息收发。 因此需要创建一个具有管理员权限的用户。
# 添加一个新用户，用户名为 root，密码为 123456
sudo rabbitmqctl add_user root 123456
# 将 root 用户标记为 administrator，使其拥有管理权限
sudo rabbitmqctl set_user_tags root administrator
# 为 root 用户设置权限，允许其在默认虚拟主机 "/" 上执行所有操作
# 参数依次为：虚拟主机、配置权限、写权限、读权限
sudo rabbitmqctl set_permissions -p / root "." "." ".*"
# RabbitMQ 自带 Web 管理界面，需要启用管理插件才能访问
sudo rabbitmq-plugins enable rabbitmq_management
我们去浏览器访问 webUI 界面，默认端口为 15672。
your_host_ip:15672
输入设置的 root 账号密码即可登录。 至此 RabbitMQ 安装成功。
AMQP-CPP 的安装
我们这里使用 AMQP-CPP 库来编写客户端程序。
# 安装 libev 开发库，这是 AMQP-CPP 依赖的网络库组件
sudo apt install libev-dev
# 从 GitHub 克隆 AMQP-CPP 库的源代码
git clone https://github.com/CopernicaMarketingSoftware/AMQP-CPP.git
# 进入克隆下来的源代码目录
cd AMCPP/
# 编译源代码，生成库文件
make
# 将编译好的库文件安装到系统目录中
sudo make install
注意：如果你在 make 的时候遇到了 SSL 版本相关的错误，表示 ssl 版本出现问题。解决方案是卸载当前的 ssl 库，重新进行修复安装，然后重新执行 make 即可。
test.cpp
// 包含 libev 事件循环库的头文件，用于事件驱动
#include <ev.h>
// 包含 AMQP-CPP 核心库，提供 AMQP 客户端功能
#include <amqpcpp.h>
// 包含 AMQP-CPP 与 libev 的集成头文件，使 AMQP 事件能够集成到 libev 循环中
#include <amqpcpp/libev.h>
// 包含标准输入输出流，用于打印信息
#include <iostream>
int main() {
// 获取默认的 libev 事件循环实例（EV_DEFAULT 是一个宏，返回默认事件循环指针）
auto *loop = EV_DEFAULT;
// 创建 AMQP 事件处理器，将 AMQP 的事件回调集成到 libev 循环中
// LibEvHandler 接受一个 libev 事件循环指针，并负责将 AMQP 底层 socket 事件注册到该循环
AMQP::LibEvHandler handler(loop);
// 创建 AMQP TCP 连接，指定 RabbitMQ 服务器地址
// 地址格式：amqp://用户名：密码@主机：端口/虚拟主机
// 这里使用 root:123456 登录本机的默认虚拟主机 "/"
AMQP::TcpConnection connection(&handler, AMQP::Address("amqp://root:[email protected]:5672/"));
// 在连接上创建通道，几乎所有 AMQP 操作（声明队列、发布、消费等）都通过通道进行
AMQP::TcpChannel channel(&connection);
// 设置通道就绪时的回调函数
// 当 TCP 连接成功建立且通道可操作时，该回调被触发
channel.onReady([&channel, &connection]() {
// 现在通道已就绪，可以执行 AMQP 命令
// 声明一个名为 "test_queue" 的队列
// 如果队列不存在，RabbitMQ 会自动创建它；如果存在，则直接使用现有队列
// declareQueue() 返回一个代表异步操作的对象，我们可以通过 onSuccess() 设置成功回调
channel.declareQueue("test_queue")
.onSuccess([&channel, &connection]() {
// 队列声明成功后的回调
// 向默认交换机（空字符串表示默认交换机）发送一条消息
// 路由键为 "test_queue"，消息将进入同名队列
// publish() 参数：交换机、路由键、消息体（可附加属性，这里省略）
channel.publish("", "test_queue", "Hello from AMQP-CPP with lambda!");
// 输出提示信息
std::cout << "[x] Sent a message to 'test_queue'" << std::endl;
// 现在开始消费队列 "test_queue"，当消息到达时自动调用回调
// consume() 返回一个代表消费操作的对象，通过 onReceived() 设置消息到达时的回调
channel.consume("test_queue")
.onReceived([&channel, &connection](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered) {
// 消息到达时的回调，参数包含消息内容、投递标签和重传标志
// 输出接收到的消息内容（message.body() 返回消息体的字符串）
std::cout << "[x] Received '" << message.body() << "'" << std::endl;
// 确认消息已被处理，告知 RabbitMQ 可以从队列中删除该消息
// 参数 deliveryTag 是服务器分配的唯一标识
channel.ack(deliveryTag);
// 关闭连接，程序将退出（事件循环会因连接关闭而结束）
connection.close();
});
})
.onError([&channel, &connection](const char *message) {
// 如果队列声明过程中发生错误，该回调被调用
std::cerr << "Queue declaration error: " << message << std::endl;
// 发生错误时也关闭连接，避免程序一直挂起
connection.close();
});
});
// 启动 libev 事件循环，程序将阻塞在这里，直到连接关闭或出现错误
// 第二个参数 0 表示正常运行，直到 ev_break 被调用
ev_run(loop, 0);
return 0;
}
Makefile
publish_and_consume : test.cpp
g++ -std=c++17 $^ -o $@ -lamqpcpp -lev
.PHONY : clean
clean : rm -rf publish_and_consume
运行结果正常，末尾乱码不影响功能。
send.cpp
#include <ev.h> // libev 事件循环库
#include <amqpcpp.h> // AMQP-CPP 核心库
#include <amqpcpp/libev.h> // AMQP-CPP 与 libev 的集成
#include <iostream> // 标准输入输出
int main() {
// 获取默认的 libev 事件循环实例
auto *loop = EV_DEFAULT;
// 创建 AMQP 事件处理器，将 AMQP 事件集成到 libev 循环中
AMQP::LibEvHandler handler(loop);
// 建立 TCP 连接（本地 RabbitMQ，默认端口 5672）
AMQP::TcpConnection connection(&handler, AMQP::Address("amqp://root:[email protected]:5672/"));
// 在连接上创建通道，几乎所有操作都在通道上进行
AMQP::TcpChannel channel(&connection);
// 设置通道就绪时的回调函数：当 TCP 连接成功且通道可使用时，执行 Lambda
channel.onReady([&channel, &connection]() {
// 声明一个名为 "hello" 的队列（如果队列不存在则自动创建）
// declareQueue 返回一个异步操作对象，通过 onSuccess 设置成功回调
channel.declareQueue("hello")
.onSuccess([&channel, &connection]() {
// 队列声明成功后，向默认交换机发送一条消息
channel.publish("", "hello", "Hello World!");
std::cout << "[x] Sent 'Hello World!'" << std::endl;
// 关闭连接，程序将退出（连接关闭会触发事件循环结束）
connection.close();
});
});
// 启动 libev 事件循环，该循环会阻塞，直到所有事件处理完毕
ev_run(loop, 0);
return 0;
}
receive.cpp
#include <ev.h> // libev 事件循环库
#include <amqpcpp.h> // AMQP-CPP 核心库
#include <amqpcpp/libev.h> // AMQP-CPP 与 libev 的集成
#include <iostream> // 标准输入输出
int main() {
// 获取默认的 libev 事件循环实例
auto *loop = EV_DEFAULT;
// 创建 AMQP 事件处理器，将 AMQP 事件集成到 libev 循环中
AMQP::LibEvHandler handler(loop);
// 建立 TCP 连接（本地 RabbitMQ，默认端口 5672）
AMQP::TcpConnection connection(&handler, AMQP::Address("amqp://root:[email protected]:5672/"));
// 在连接上创建通道，几乎所有操作都在通道上进行
AMQP::TcpChannel channel(&connection);
// 设置通道就绪时的回调函数：当 TCP 连接成功且通道可使用时，执行 Lambda
channel.onReady([&channel, &connection]() {
// 声明一个名为 "hello" 的队列（如果队列不存在则自动创建）
// declareQueue 返回一个异步操作对象，通过 onSuccess 设置成功回调
channel.declareQueue("hello")
.onSuccess([&channel, &connection]() {
// 开始消费队列 "hello"，当有消息到达时，调用 Lambda 处理
channel.consume("hello")
.onReceived([&channel, &connection](const AMQP::Message& message, uint64_t deliveryTag, bool redelivered) {
// 输出接收到的消息内容
std::cout << "[x] Received '" << message.body() << "'" << std::endl;
// 确认消息已被处理，告知 RabbitMQ 可以删除该消息
channel.ack(deliveryTag);
// 关闭连接，程序将退出（连接关闭会触发事件循环结束）
connection.close();
});
});
});
// 启动 libev 事件循环，该循环会阻塞，直到所有事件处理完毕
ev_run(loop, 0);
return 0;
}
Makefile
all : send receive
send : send.cpp
g++ -std=c++17 $^ -o $@ -lamqpcpp -lev
receive : receive.cpp
g++ -std=c++17 $^ -o $@ -lamqpcpp -lev
.PHONY : clean
clean : rm -rf send receive
由于使用了 std::cout 来打印二进制数据，终端显示时可能会出现乱码，只要不去打印即可，不影响功能。
publish.cc
// 包含 libev 事件循环库的头文件，用于事件驱动
#include <ev.h>
// 包含 AMQP-CPP 核心库，提供 AMQP 客户端功能
#include <amqpcpp.h>
// 包含 AMQP-CPP 与 libev 的集成头文件，使 AMQP 事件能够集成到 libev 循环中
#include <amqpcpp/libev.h>
// 包含 OpenSSL 的 SSL 功能头文件（本例中未直接使用，可能为后续扩展预留）
#include <openssl/ssl.h>
// 包含 OpenSSL 的版本信息头文件
#include <openssl/opensslv.h>
int main() {
// 1. 获取默认的 libev 事件循环实例（EV_DEFAULT 宏返回默认事件循环指针）
auto *loop = EV_DEFAULT;
// 2. 创建 AMQP 事件处理器，将 AMQP 的事件回调集成到 libev 循环中
AMQP::LibEvHandler handler(loop);
// 2.5. 创建 AMQP 服务器地址对象，指定 RabbitMQ 连接信息（用户名：密码@主机：端口/虚拟主机）
AMQP::Address address("amqp://root:[email protected]:5672/");
// 创建 TCP 连接对象，传入事件处理器和地址，负责底层网络通信
AMQP::TcpConnection connection(&handler, address);
// 3. 在连接上创建通道，几乎所有 AMQP 操作都通过通道进行
AMQP::TcpChannel channel(&connection);
// 4. 声明一个直连类型的交换机（Direct Exchange），名称为 "test-exchange"
// declareExchange 返回一个 Deferred 对象，用于设置成功/失败回调
channel.declareExchange("test-exchange", AMQP::ExchangeType::direct)
.onError([](const char *message) {
// 如果声明交换机失败，输出错误信息并退出程序
std::cout << "声明交换机失败：" << message << std::endl;
exit(0);
})
.onSuccess([](){
// 声明交换机成功时的提示
std::cout << "test-exchange 交换机创建成功！" << std::endl;
});
// 5. 声明一个队列，名称为 "test-queue"
channel.declareQueue("test-queue")
.onError([](const char *message) {
// 如果声明队列失败，输出错误信息并退出
std::cout << "声明队列失败：" << message << std::endl;
exit(0);
})
.onSuccess([](){
// 声明队列成功时的提示
std::cout << "test-queue 队列创建成功！" << std::endl;
});
// 6. 将交换机和队列通过路由键 "test-queue-key" 进行绑定
channel.bindQueue("test-exchange", "test-queue", "test-queue-key")
.onError([](const char *message) {
// 如果绑定失败，输出错误信息并退出
std::cout << "test-exchange - test-queue 绑定失败：" << message << std::endl;
exit(0);
})
.onSuccess([](){
// 绑定成功时的提示
std::cout << "test-exchange - test-queue 绑定成功！" << std::endl;
});
// 7. 向交换机发布消息，循环发送 10 条消息
for (int i = 0; i < 10; i++) {
// 构造消息内容，包含循环计数
std::string msg = "Hello Bite-" + std::to_string(i);
// 发布消息到指定交换机，使用路由键 "test-queue-key"，消息将路由到绑定的队列
bool ret = channel.publish("test-exchange", "test-queue-key", msg);
if (ret == false) {
// 如果 publish 返回 false，表示消息未能立即写入发送缓冲区（可能连接未就绪）
std::cout << "publish 失败！\n";
}
}
// 8. 启动 libev 事件循环，程序将阻塞在这里，直到连接关闭或出现错误
// 注意：由于没有显式关闭连接，程序将一直运行，等待事件发生
ev_run(loop, 0);
return 0;
}
consume.cc
// 包含 libev 事件循环库的头文件，用于事件驱动
#include <ev.h>
// 包含 AMQP-CPP 核心库，提供 AMQP 客户端功能
#include <amqpcpp.h>
// 包含 AMQP-CPP 与 libev 的集成头文件，使 AMQP 事件能够集成到 libev 循环中
#include <amqpcpp/libev.h>
// 包含 OpenSSL 的 SSL 功能头文件（本例中未使用，但可能为后续扩展预留）
#include <openssl/ssl.h>
// 包含 OpenSSL 的版本信息头文件
#include <openssl/opensslv.h>
// 消息回调处理函数的实现，当从队列中收到消息时被调用
void MessageCb(AMQP::TcpChannel *channel, const AMQP::Message &message, uint64_t deliveryTag, bool redelivered) {
// 将消息体（二进制数据）构造为 std::string 以便打印
std::string msg;
msg.assign(message.body(), message.bodySize());
// 输出消息内容到标准输出
std::cout << msg << std::endl;
// 确认消息已被处理，告知 RabbitMQ 可以删除该消息
channel->ack(deliveryTag);
}
int main() {
// 1. 获取默认的 libev 事件循环实例（EV_DEFAULT 宏返回默认事件循环指针）
auto *loop = EV_DEFAULT;
// 2. 创建 AMQP 事件处理器，将 AMQP 的事件回调集成到 libev 循环中
AMQP::LibEvHandler handler(loop);
// 2.5. 创建 AMQP 服务器地址对象，指定 RabbitMQ 连接信息（用户名：密码@主机：端口/虚拟主机）
AMQP::Address address("amqp://root:[email protected]:5672/");
// 创建 TCP 连接对象，传入事件处理器和地址，负责底层网络通信
AMQP::TcpConnection connection(&handler, address);
// 3. 在连接上创建通道，几乎所有 AMQP 操作都通过通道进行
AMQP::TcpChannel channel(&connection);
// 4. 声明一个直连类型的交换机（Direct Exchange），名称为 "test-exchange"
// declareExchange 返回一个 Deferred 对象，用于设置成功/失败回调
channel.declareExchange("test-exchange", AMQP::ExchangeType::direct)
.onError([](const char *message) {
// 如果声明交换机失败，输出错误信息并退出程序
std::cout << "声明交换机失败：" << message << std::endl;
exit(0);
})
.onSuccess([](){
// 声明交换机成功时的提示
std::cout << "test-exchange 交换机创建成功！" << std::endl;
});
// 5. 声明一个队列，名称为 "test-queue"
channel.declareQueue("test-queue")
.onError([](const char *message) {
// 如果声明队列失败，输出错误信息并退出
std::cout << "声明队列失败：" << message << std::endl;
exit(0);
})
.onSuccess([](){
// 声明队列成功时的提示
std::cout << "test-queue 队列创建成功！" << std::endl;
});
// 6. 将交换机和队列通过路由键 "test-queue-key" 进行绑定
channel.bindQueue("test-exchange", "test-queue", "test-queue-key")
.onError([](const char *message) {
// 如果绑定失败，输出错误信息并退出
std::cout << "test-exchange - test-queue 绑定失败：" << message << std::endl;
exit(0);
})
.onSuccess([](){
// 绑定成功时的提示
std::cout << "test-exchange - test-queue 绑定成功！" << std::endl;
});
// 7. 订阅队列消息 -- 设置消息处理回调函数
// 使用 std::bind 将 MessageCb 函数与 channel 指针绑定，并预留占位符供实际参数使用
auto callback = std::bind(MessageCb, &channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
// 开始消费队列 "test-queue"，消费者标签为 "consume-tag"，返回 DeferredConsumer 对象
channel.consume("test-queue", "consume-tag")
.onReceived(callback) // 设置消息到达时的回调
.onError([](const char *message){
// 如果订阅过程中发生错误，输出错误信息并退出
std::cout << "订阅 test-queue 队列消息失败:" << message << std::endl;
exit(0);
});
// 8. 启动 libev 事件循环，程序将阻塞在这里，直到连接关闭或出现错误
ev_run(loop, 0);
return 0;
}
Makefile
all : publish consume
publish : publish.cc
g++ -std=c++17 $^ -o $@ -lamqpcpp -lev
consume : consume.cc
g++ -std=c++17 $^ -o $@ -lamqpcpp -lev
.PHONY : clean
clean : rm publish consume
运行结果符合预期。
基于上述示例，我们可以进一步对 AMQP-CPP 接口进行二次封装，以更方便快捷地实现与 RabbitMQ 的沟通。
class MQClient {
public:
// 定义消息回调函数类型，接收消息指针和大小
using MessageCallback = std::function<void(const char*, size_t)>;
// 智能指针类型别名，便于管理 MQClient 对象的生命周期
using ptr = std::shared_ptr<MQClient>;
/**
* @brief 构造函数，初始化与 RabbitMQ 的连接并启动事件循环线程。
* @param user RabbitMQ 用户名
* @param passwd RabbitMQ 密码
* @param host RabbitMQ 服务器地址（格式：ip:port，例如 "127.0.0.1:5672"）
*/
MQClient(const std::string &user, const std::string passwd, const std::string host) {
// 获取默认的 libev 事件循环
_loop = EV_DEFAULT;
// 创建 AMQP 的 libev 事件处理器，将其与事件循环关联
_handler = std::make_unique<AMQP::LibEvHandler>(_loop);
// 构造 AMQP 连接 URL，格式：amqp://user:password@host/
std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
AMQP::Address address(url); // 解析地址
// 创建 TCP 连接对象，使用 handler 处理事件，连接到指定地址
_connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
// 创建通道，所有 AMQP 操作都通过通道进行
_channel = std::make_unique<AMQP::TcpChannel>(_connection.get());
// 启动一个线程专门运行 libev 事件循环，使网络事件能够异步处理
_loop_thread = std::thread([this]() {
ev_run(_loop, 0); // 开始事件循环，0 表示一直运行直到被停止
});
}
// ... 省略部分代码 ...
private:
struct ev_async _async_watcher; // libev 异步 watcher，用于退出事件循环
struct ev_loop *_loop; // libev 事件循环指针
std::unique_ptr<AMQP::LibEvHandler> _handler; // AMQP 事件处理器，连接 libev 和 AMQP-CPP
std::unique_ptr<AMQP::TcpConnection> _connection; // AMQP TCP 连接
std::unique_ptr<AMQP::TcpChannel> _channel; // AMQP 通道，用于执行命令
std::thread _loop_thread; // 运行事件循环的线程
};
针对构造函数和成员变量的设计，基本逻辑与示例一致。
析构函数涉及 libev 事件循环的异步通知机制，用于在多线程环境中安全地停止事件循环：
~MQClient() {
// 初始化一个异步 watcher，用于通知事件循环退出
ev_async_init(&_async_watcher, watcher_callback);
// 在事件循环中启动该 watcher
ev_async_start(_loop, &_async_watcher);
// 发送异步信号，触发 watcher_callback，从而停止事件循环
ev_async_send(_loop, &_async_watcher);
// 等待事件循环线程结束
_loop_thread.join();
// 清空事件循环指针
_loop = nullptr;
}
static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents) {
ev_break(loop, EVBREAK_ALL); // 停止所有事件循环
}
这段代码通过 ev_async 机制实现了跨线程通知，确保主线程可以安全地终止事件循环线程。
创建交换机，队列，并将它们进行绑定
/**
* @brief 声明交换机、队列并进行绑定。
* @param exchange 交换机名称
* @param queue 队列名称
* @param routing_key 绑定时的路由键，默认 "routing_key"
* @param echange_type 交换机类型，默认 direct
*/
void declareComponents(const std::string &exchange, const std::string &queue, const std::string &routing_key = "routing_key", AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct)
{
// 声明交换机
_channel->declareExchange(exchange, echange_type)
.onError([](const char *message) {
LOG_ERROR("声明交换机失败：{}", message);
exit(0);
})
.onSuccess([exchange]() {
LOG_DEBUG("{} 交换机创建成功！", exchange);
});
// 声明队列
_channel->declareQueue(queue)
.onError([](const char *message) {
LOG_ERROR("声明队列失败：{}", message);
exit(0);
})
.onSuccess([queue]() {
LOG_DEBUG("{} 队列创建成功！", queue);
});
// 将队列绑定到交换机，使用指定的路由键
_channel->bindQueue(exchange, queue, routing_key)
.onError([exchange, queue](const char *message) {
LOG_ERROR("{} - {} 绑定失败：", exchange, queue);
exit(0);
})
.onSuccess([exchange, queue, routing_key]() {
LOG_DEBUG("{} - {} - {} 绑定成功！", exchange, queue, routing_key);
});
}
发布消息
bool publish(const std::string &exchange, const std::string &msg, const std::string &routing_key = "routing_key") {
LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routing_key);
// 调用 channel 的 publish 方法，将消息发布到指定交换机和路由键
bool ret = _channel->publish(exchange, routing_key, msg);
if (ret == false) {
LOG_ERROR("{} 发布消息失败：", exchange);
return false;
}
return true;
}
消费消息
void consume(const std::string &queue, const MessageCallback &cb) {
LOG_DEBUG("开始订阅 {} 队列消息！", queue);
// 调用 channel 的 consume 方法，指定消费者标签 "consume-tag"
_channel->consume(queue, "consume-tag")
.onReceived([this, cb](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered) {
// 当收到消息时，调用用户提供的回调函数
cb(message.body(), message.bodySize());
// 确认消息已被处理，以便 RabbitMQ 可以将其从队列中移除
_channel->ack(deliveryTag);
})
.onError([queue](const char *message) {
LOG_ERROR("订阅 {} 队列消息失败：{}", queue, message);
exit(0);
});
}
完整代码
#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include <thread>
#include <memory>
#include "logger.hpp"
namespace IMS {
class MQClient {
public:
using MessageCallback = std::function<void(const char*, size_t)>;
using ptr = std::shared_ptr<MQClient>;
MQClient(const std::string &user, const std::string passwd, const std::string host) {
_loop = EV_DEFAULT;
_handler = std::make_unique<AMQP::LibEvHandler>(_loop);
std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
AMQP::Address address(url);
_connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
_channel = std::make_unique<AMQP::TcpChannel>(_connection.get());
_loop_thread = std::thread([this]() {
ev_run(_loop, 0);
});
}
~MQClient() {
ev_async_init(&_async_watcher, watcher_callback);
ev_async_start(_loop, &_async_watcher);
ev_async_send(_loop, &_async_watcher);
_loop_thread.join();
_loop = nullptr;
}
void declareComponents(const std::string &exchange, const std::string &queue, const std::string &routing_key = "routing_key", AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct) {
_channel->declareExchange(exchange, echange_type)
.onError([](const char *message) { LOG_ERROR("声明交换机失败：{}", message); exit(0); })
.onSuccess([exchange]() { LOG_DEBUG("{} 交换机创建成功！", exchange); });
_channel->declareQueue(queue)
.onError([](const char *message) { LOG_ERROR("声明队列失败：{}", message); exit(0); })
.onSuccess([queue]() { LOG_DEBUG("{} 队列创建成功！", queue); });
_channel->bindQueue(exchange, queue, routing_key)
.onError([exchange, queue](const char *message) { LOG_ERROR("{} - {} 绑定失败：", exchange, queue); exit(0); })
.onSuccess([exchange, queue, routing_key]() { LOG_DEBUG("{} - {} - {} 绑定成功！", exchange, queue, routing_key); });
}
bool publish(const std::string &exchange, const std::string &msg, const std::string &routing_key = "routing_key") {
LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routing_key);
bool ret = _channel->publish(exchange, routing_key, msg);
if (ret == false) {
LOG_ERROR("{} 发布消息失败：", exchange);
return false;
}
return true;
}
void consume(const std::string &queue, const MessageCallback &cb) {
LOG_DEBUG("开始订阅 {} 队列消息！", queue);
_channel->consume(queue, "consume-tag")
.onReceived([this, cb](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered) {
cb(message.body(), message.bodySize());
_channel->ack(deliveryTag);
})
.onError([queue](const char *message) { LOG_ERROR("订阅 {} 队列消息失败：{}", queue, message); exit(0); });
}
private:
static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents) {
ev_break(loop, EVBREAK_ALL);
}
struct ev_async _async_watcher;
struct ev_loop *_loop;
std::unique_ptr<AMQP::LibEvHandler> _handler;
std::unique_ptr<AMQP::TcpConnection> _connection;
std::unique_ptr<AMQP::TcpChannel> _channel;
std::thread _loop_thread;
};
}
我们先来讲讲我们的消息队列是怎么进行路由匹配的。
最关键的就是下面这 2 个东西：
Routing Key（路由键）
Binding Key（绑定键）
它们是怎么进行工作的？
步骤 1：生产者发送 生产者应用创建一个消息，为其指定一个路由键（Routing Key），然后将其发送到 Broker 上一个已知的 Exchange。
步骤 2：交换机路由 Exchange 收到消息后，会提取消息中的 Routing Key。然后，它查看所有绑定（Binding）到自身的规则列表。
步骤 3：绑定匹配与投递 对于每一条绑定规则，Exchange 会根据自身的类型，将消息的 Routing Key 与绑定的 Binding Key 进行匹配。
步骤 4：队列存储与消费 消息被投递到队列后，便存储在队列中。消费者应用从自己订阅的队列中获取消息进行处理。
那么它们怎么进行匹配呢？
交换机的类型决定了 Binding Key 和 Routing Key 的匹配算法。
1. 直连交换机（Direct）
示例：
用途：点对点精确路由，常用于任务分发。
2. 主题交换机（Topic）
匹配规则：通配符模式匹配。这是最灵活、最常用的路由方式。
Binding Key：是一个用点号.分隔的单词组成的模式，支持两个通配符：
Routing Key：也是一个用点号.分隔的单词组成的字符串（不能包含通配符）。
示例：
用途：基于多重标准的灵活消息路由，如日志系统、事件通知系统。
3. 扇出交换机（Fanout）
那么在我们这里我们只会去测试这个 direct 交换机。
示例 1
direct.cpp
#include <iostream>
#include <atomic>
#include <thread>
#include <chrono>
#include "../../common/rabbitmq.hpp"
int main() {
IMS::init_logger(false, "", 0);
// 1. 创建 MQClient 对象，连接到本地 RabbitMQ
IMS::MQClient client("root", "123456", "127.0.0.1:5672");
// 2. 声明组件：交换机 "test-exchange"、队列 "test-queue"、绑定键 "test-key"
client.declareComponents("test-exchange", "test-queue", "test-key");
// 3. 设置消费回调：当收到队列 "test-queue" 的消息时打印内容并通知主线程
std::atomic<bool> messageReceived(false);
client.consume("test-queue", [&messageReceived](const char* data, size_t size) {
std::string msg(data, size);
std::cout << "Received message: " << msg << std::endl;
messageReceived = true;
});
// 4. 短暂等待
std::this_thread::sleep_for(std::chrono::milliseconds(500));
// 5. 发布一条消息到交换机 "test-exchange"，路由键 "test-key"
bool pubSuccess = client.publish("test-exchange", "Hello, RabbitMQ!", "test-key");
if (!pubSuccess) {
std::cerr << "Failed to publish message!" << std::endl;
return 1;
}
std::cout << "Message published." << std::endl;
// 6. 等待消息被消费，设置超时避免无限等待
std::this_thread::sleep_for(std::chrono::seconds(3));
// 7. 检查结果
if (messageReceived) {
std::cout << "Test passed: message received successfully." << std::endl;
return 0;
} else {
std::cerr << "Test failed: timeout waiting for message." << std::endl;
return 1;
}
}
运行日志：
[default-logger][20:35:48][396036][debug ][../../common/rabbitmq.hpp:150] 开始订阅 test-queue 队列消息！
[default-logger][20:35:48][396038][debug ][../../common/rabbitmq.hpp:93] test-exchange 交换机创建成功！
[default-logger][20:35:48][396038][debug ][../../common/rabbitmq.hpp:105] test-queue 队列创建成功！
[default-logger][20:35:48][396038][debug ][../../common/rabbitmq.hpp:117] test-exchange - test-queue - test-key 绑定成功！
[default-logger][20:35:49][396036][debug ][../../common/rabbitmq.hpp:133] 向交换机 test-exchange-test-key 发布消息！
Message published.
Received message: Hello, RabbitMQ!
示例 2
publish.cc
#include "../../../../common/rabbitmq.hpp"
#include "../../../../common/logger.hpp"
#include <gflags/gflags.h>
DEFINE_string(user, "root", "rabbitmq 访问用户名");
DEFINE_string(pswd, "123456", "rabbitmq 访问密码");
DEFINE_string(host, "127.0.0.1:5672", "rabbitmq 服务器地址信息 host:port");
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试；true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
int main(int argc, char *argv[]) {
google::ParseCommandLineFlags(&argc, &argv, true);
IMS::init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
IMS::MQClient client(FLAGS_user, FLAGS_pswd, FLAGS_host);
client.declareComponents("test-exchange", "test-queue");
for (int i = 0; i < 10; i++) {
std::string msg = "Hello Bite-" + std::to_string(i);
bool ret = client.publish("test-exchange", msg);
if (ret == false) {
std::cout << "publish 失败！\n";
}
}
std::this_thread::sleep_for(std::chrono::seconds(3));
return 0;
}
consume.cc
#include "../../../../common/rabbitmq.hpp"
#include "../../../../common/logger.hpp"
#include <gflags/gflags.h>
DEFINE_string(user, "root", "rabbitmq 访问用户名");
DEFINE_string(pswd, "123456", "rabbitmq 访问密码");
DEFINE_string(host, "127.0.0.1:5672", "rabbitmq 服务器地址信息 host:port");
DEFINE_bool(run_mode, false, "程序的运行模式，false-调试；true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");
void callback(const char *body, size_t sz) {
std::string msg;
msg.assign(body, sz);
std::cout << msg << std::endl;
}
int main(int argc, char *argv[]) {
google::ParseCommandLineFlags(&argc, &argv, true);
IMS::init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);
IMS::MQClient client(FLAGS_user, FLAGS_pswd, FLAGS_host);
client.declareComponents("test-exchange", "test-queue");
client.consume("test-queue", callback);
std::this_thread::sleep_for(std::chrono::seconds(60));
return 0;
}
Makefile
all : publish consume
publish : publish.cc
g++ -g -std=c++17 $^ -o $@ -lamqpcpp -lev -lfmt -lspdlog -lgflags
consume : consume.cc
g++ -g -std=c++17 $^ -o $@ -lamqpcpp -lev -lfmt -lspdlog -lgflags
.PHONY : clean
clean : rm publish consume
运行成功。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online
将JSON字符串修饰为友好的可读格式。 在线工具，JSON美化和格式化在线工具，online