目录
一.广播交换模式下的测试
1.1.生产者消费者代码编写
1.2.测试
二.直接交换模式下的测试
2.1.生产者消费者代码编写
2.2.测试
三.主题交换模式下的测试
3.1.生产者消费者代码编写
3.2.测试
搭建客户端
- 发布消息的生产者客户端
- 订阅消息的消费者客户端
思想
- 必须要有一个生产者客户端
- 声明一个交换机
- 声明一个队列1，binding_key=queue1
- 声明一个队列2，binding_key=news.music.#
- 将两个队列和交换机绑定起来
- 搭建两个消费者客户端，分别订阅一个队列的消息
测试
- 第一次，将交换机类型设置为广播模式，理论结果两个消费者客户端都能拿到消息
- 第二次，将交换机类型设置为直接交换模式，routing_key=queue1 ,理论结果，只有订阅了队列1消息的客户端能拿到消息
- 第三次，将交换机类型设置为主题交换，routing_key=news.music.pop，理论结果只有订阅了队列2消息的客户端能拿到消息
一.广播交换模式下的测试
1.1.生产者消费者代码编写
生产者客户端实现
#include "connection.hpp" int main() { // 1. 创建异步工作线程对象，用于处理网络IO mymq::AsyncWorker::ptr awp = std::make_shared<mymq::AsyncWorker>(); // 2. 建立与RabbitMQ服务器的连接 mymq::Connection::ptr conn = std::make_shared<mymq::Connection>("127.0.0.1", 8085, awp); // 3. 通过连接创建信道 mymq::Channel::ptr channel = conn->openChannel(); // 4. 声明交换机和队列，并建立绑定关系 // 使用空的属性映射（不设置额外参数） google::protobuf::Map<std::string, std::string> tmp_map; // 4.1 声明一个名为"exchange1"的广播交换机（ExchangeType::FANOUT） // 参数：交换机名称、类型、是否持久化、是否自动删除、额外属性 channel->declareExchange("exchange1", mymq::ExchangeType::FANOUT, true, false, tmp_map); // 4.2 声明一个持久化队列"queue1" // 参数：队列名称、是否持久化、是否排他、是否自动删除、额外属性 channel->declareQueue("queue1", true, false, false, tmp_map); // 4.3 声明一个持久化队列"queue2" channel->declareQueue("queue2", true, false, false, tmp_map); // 4.4 将队列"queue1"绑定到交换机"exchange1" channel->queueBind("exchange1", "queue1", "queue1"); // 4.5 将队列"queue2"绑定到交换机"exchange1" channel->queueBind("exchange1", "queue2", "news.music.#"); // 5. 发布消息到交换机 // 发送10条消息，所有绑定队列都会收到 for (int i = 0; i < 10; i++) { // 发布消息：交换机名称、消息属性、消息内容 channel->basicPublish("exchange1", nullptr, "Hello World-" + std::to_string(i)); } // 6. 关闭信道 conn->closeChannel(channel); return 0; }消费者
#include "connection.hpp" #include <iostream> #include <thread> #include <chrono> #include <functional> // 消息消费回调函数 // 参数： // channel: 消息通道指针，用于发送确认消息等操作 // consumer_tag: 消费者标签，标识不同的消费者 // bp: 消息基本属性，包含消息ID、路由键等信息 // body: 消息体内容 void cb(mymq::Channel::ptr &channel, const std::string consumer_tag, const mymq::BasicProperties *bp, const std::string &body) { // 打印消费者信息和接收到的消息内容 std::cout << consumer_tag << "消费了消息：" << body << std::endl; // 发送消息确认，告诉RabbitMQ消息已被成功处理 // 参数：消息唯一ID channel->basicAck(bp->id()); } int main(int argc, char *argv[]) { // 检查命令行参数：需要指定要消费的队列名称 if (argc != 2) { std::cout << "使用方法: ./consume_client <队列名称>\n"; std::cout << "示例: ./consume_client queue1\n"; return -1; } // 1. 创建异步工作线程对象，用于处理网络IO和消息回调 mymq::AsyncWorker::ptr awp = std::make_shared<mymq::AsyncWorker>(); // 2. 建立与RabbitMQ服务器的连接 // 参数：服务器地址、端口号、异步工作线程 mymq::Connection::ptr conn = std::make_shared<mymq::Connection>("127.0.0.1", 8085, awp); // 3. 通过连接创建信道，信道是执行AMQP操作的通道 mymq::Channel::ptr channel = conn->openChannel(); // 4. 声明交换机和队列，并建立绑定关系 // 使用空的属性映射（不设置额外参数） google::protobuf::Map<std::string, std::string> tmp_map; // 4.1 声明一个名为"exchange1"的直接交换机（ExchangeType::FANOUT） // 参数：交换机名称、类型、是否持久化、是否自动删除、额外属性 channel->declareExchange("exchange1", mymq::ExchangeType::FANOUT, true, false, tmp_map); // 4.2 声明一个持久化队列"queue1" // 参数：队列名称、是否持久化、是否排他、是否自动删除、额外属性 channel->declareQueue("queue1", true, false, false, tmp_map); // 4.3 声明一个持久化队列"queue2" channel->declareQueue("queue2", true, false, false, tmp_map); // 4.4 将队列"queue1"绑定到交换机"exchange1"，绑定键为"queue1" // 参数：交换机名称、队列名称、绑定键 channel->queueBind("exchange1", "queue1", "queue1"); // 4.5 将队列"queue2"绑定到交换机"exchange1"，绑定键为"news.music.#" // "#"是通配符，表示匹配多个单词 channel->queueBind("exchange1", "queue2", "news.music.#"); // 5. 使用std::bind创建回调函数适配器 // std::bind将回调函数cb与参数绑定，其中channel作为第一个参数固定传入 // std::placeholders::_1、_2、_3代表回调函数cb的consumer_tag、bp、body参数 auto functor = std::bind(cb, channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3); // 6. 开始消费指定队列的消息 // 参数： // consumer1: 消费者标签，用于标识此消费者 // argv[1]: 队列名称，从命令行参数获取 // false: 是否自动确认消息，false表示需要手动确认（basicAck） // functor: 消息到达时的回调函数 channel->basicConsume("consumer1", argv[1], false, functor); // 7. 保持程序运行，等待消息 // 使用无限循环，每3秒休眠一次，避免CPU占用过高 while(1) { std::this_thread::sleep_for(std::chrono::seconds(3)); } // 8. 关闭信道（这里不会被执行，因为程序一直循环等待消息） conn->closeChannel(channel); return 0; }makefile
all: publish_client consume_client publish_client: publish_client.cc ../third/include/muduo/protobuf/codec.cc ../mqcommon/msg.pb.cc ../mqcommon/proto.pb.cc g++ $^ -o $@ -I ../third/include -L ../third/lib -lmuduo_net -lmuduo_base -lpthread -lprotobuf -lz -lsqlite3 consume_client: consume_client.cc ../third/include/muduo/protobuf/codec.cc ../mqcommon/msg.pb.cc ../mqcommon/proto.pb.cc g++ $^ -o $@ -I ../third/include -L ../third/lib -lmuduo_net -lmuduo_base -lpthread -lprotobuf -lz -lsqlite3 .PHONY: clean clean: rm -rf publish_client consume_client1.2.测试
没有问题，接下来我们就进入我们的测试环节
首先我们先启动服务器
服务器已经运行起来了
我们先运行消费者客户端
./consume_client queue1换一个终端
./consume_client queue2换一个终端
./publish_client运行之后，我们发现绑定了queue1或者queue2的都收到了消息
二.直接交换模式下的测试
2.1.生产者消费者代码编写
生产者
#include "connection.hpp" int main() { // 1. 创建异步工作线程对象，用于处理网络IO mymq::AsyncWorker::ptr awp = std::make_shared<mymq::AsyncWorker>(); // 2. 建立与RabbitMQ服务器的连接 mymq::Connection::ptr conn = std::make_shared<mymq::Connection>("127.0.0.1", 8085, awp); // 3. 通过连接创建信道 mymq::Channel::ptr channel = conn->openChannel(); // 4. 声明交换机和队列，并建立绑定关系 // 使用空的属性映射（不设置额外参数） google::protobuf::Map<std::string, std::string> tmp_map; // 4.1 声明一个名为"exchange1"的直接交换机 // 参数：交换机名称、类型、是否持久化、是否自动删除、额外属性 channel->declareExchange("exchange1", mymq::ExchangeType::DIRECT, true, false, tmp_map); // 4.2 声明一个持久化队列"queue1" // 参数：队列名称、是否持久化、是否排他、是否自动删除、额外属性 channel->declareQueue("queue1", true, false, false, tmp_map); // 4.3 声明一个持久化队列"queue2" channel->declareQueue("queue2", true, false, false, tmp_map); // 4.4 将队列"queue1"绑定到交换机"exchange1" channel->queueBind("exchange1", "queue1", "queue1"); // 4.5 将队列"queue2"绑定到交换机"exchange1" channel->queueBind("exchange1", "queue2", "news.music.#"); // 5. 发布消息到交换机 // 发送4条消息，只有queue1能收到 for (int i = 0; i < 4; i++) { mymq::BasicProperties bp; bp.set_id(mymq::UUIDHelper::uuid()); // 设置消息唯一ID bp.set_delivery_mode(mymq::DeliveryMode::DURABLE); // 设置消息持久化 bp.set_routing_key("queue1"); // 设置路由键 // 发布消息：交换机名称、消息属性、消息内容 channel->basicPublish("exchange1", &bp, "Hello World-" + std::to_string(i)); } // 6. 关闭信道 conn->closeChannel(channel); return 0; }消费者
#include "connection.hpp" #include <iostream> #include <thread> #include <chrono> #include <functional> // 消息消费回调函数 // 参数： // channel: 消息通道指针，用于发送确认消息等操作 // consumer_tag: 消费者标签，标识不同的消费者 // bp: 消息基本属性，包含消息ID、路由键等信息 // body: 消息体内容 void cb(mymq::Channel::ptr &channel, const std::string consumer_tag, const mymq::BasicProperties *bp, const std::string &body) { // 打印消费者信息和接收到的消息内容 std::cout << consumer_tag << "消费了消息：" << body << std::endl; // 发送消息确认，告诉RabbitMQ消息已被成功处理 // 参数：消息唯一ID channel->basicAck(bp->id()); } int main(int argc, char *argv[]) { // 检查命令行参数：需要指定要消费的队列名称 if (argc != 2) { std::cout << "使用方法: ./consume_client <队列名称>\n"; std::cout << "示例: ./consume_client queue1\n"; return -1; } // 1. 创建异步工作线程对象，用于处理网络IO和消息回调 mymq::AsyncWorker::ptr awp = std::make_shared<mymq::AsyncWorker>(); // 2. 建立与RabbitMQ服务器的连接 // 参数：服务器地址、端口号、异步工作线程 mymq::Connection::ptr conn = std::make_shared<mymq::Connection>("127.0.0.1", 8085, awp); // 3. 通过连接创建信道，信道是执行AMQP操作的通道 mymq::Channel::ptr channel = conn->openChannel(); // 4. 声明交换机和队列，并建立绑定关系 // 使用空的属性映射（不设置额外参数） google::protobuf::Map<std::string, std::string> tmp_map; // 4.1 声明一个名为"exchange1"的直接交换机 // 参数：交换机名称、类型、是否持久化、是否自动删除、额外属性 channel->declareExchange("exchange1", mymq::ExchangeType::DIRECT, true, false, tmp_map); // 4.2 声明一个持久化队列"queue1" // 参数：队列名称、是否持久化、是否排他、是否自动删除、额外属性 channel->declareQueue("queue1", true, false, false, tmp_map); // 4.3 声明一个持久化队列"queue2" channel->declareQueue("queue2", true, false, false, tmp_map); // 4.4 将队列"queue1"绑定到交换机"exchange1"，绑定键为"queue1" // 参数：交换机名称、队列名称、绑定键 channel->queueBind("exchange1", "queue1", "queue1"); // 4.5 将队列"queue2"绑定到交换机"exchange1"，绑定键为"news.music.#" // "#"是通配符，表示匹配多个单词 channel->queueBind("exchange1", "queue2", "news.music.#"); // 5. 使用std::bind创建回调函数适配器 // std::bind将回调函数cb与参数绑定，其中channel作为第一个参数固定传入 // std::placeholders::_1、_2、_3代表回调函数cb的consumer_tag、bp、body参数 auto functor = std::bind(cb, channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3); // 6. 开始消费指定队列的消息 // 参数： // consumer1: 消费者标签，用于标识此消费者 // argv[1]: 队列名称，从命令行参数获取 // false: 是否自动确认消息，false表示需要手动确认（basicAck） // functor: 消息到达时的回调函数 channel->basicConsume("consumer1", argv[1], false, functor); // 7. 保持程序运行，等待消息 // 使用无限循环，每3秒休眠一次，避免CPU占用过高 while(1) { std::this_thread::sleep_for(std::chrono::seconds(3)); } // 8. 关闭信道（这里不会被执行，因为程序一直循环等待消息） conn->closeChannel(channel); return 0; }编译还是和上面一样
2.2.测试
注意：每次测试之前都需要删除/data目录，这里面存储的是持久化的绑定信息，我们需要将它们删除
没有问题，接下来我们就进入我们的测试环节
首先我们先启动服务器
服务器已经运行起来了
我们先运行消费者客户端
./consume_client queue1换一个终端
./consume_client queue2换一个终端
./publish_client运行之后，我们发现绑定了queue1的消费者收到了消息
但是绑定queue2的消费者没有收到消息
三.主题交换模式下的测试
3.1.生产者消费者代码编写
生产者客户端实现
#include "connection.hpp" int main() { // 1. 创建异步工作线程对象，用于处理网络IO mymq::AsyncWorker::ptr awp = std::make_shared<mymq::AsyncWorker>(); // 2. 建立与RabbitMQ服务器的连接 // 参数：服务器地址、端口号、异步工作线程 mymq::Connection::ptr conn = std::make_shared<mymq::Connection>("127.0.0.1", 8085, awp); // 3. 通过连接创建信道 mymq::Channel::ptr channel = conn->openChannel(); // 4. 声明交换机和队列，并建立绑定关系 // 使用空的属性映射（不设置额外参数） google::protobuf::Map<std::string, std::string> tmp_map; // 4.1 声明一个名为"exchange1"的主题交换机（ExchangeType::TOPIC） // 参数：交换机名称、类型、是否持久化、是否自动删除、额外属性 channel->declareExchange("exchange1", mymq::ExchangeType::TOPIC, true, false, tmp_map); // 4.2 声明一个持久化队列"queue1" // 参数：队列名称、是否持久化、是否排他、是否自动删除、额外属性 channel->declareQueue("queue1", true, false, false, tmp_map); // 4.3 声明一个持久化队列"queue2" channel->declareQueue("queue2", true, false, false, tmp_map); // 4.4 将队列"queue1"绑定到交换机"exchange1"，绑定键为"queue1" // 参数：交换机名称、队列名称、绑定键 channel->queueBind("exchange1", "queue1", "queue1"); // 4.5 将队列"queue2"绑定到交换机"exchange1"，绑定键为"news.music.#" // "#"是通配符，表示匹配多个单词 channel->queueBind("exchange1", "queue2", "news.music.#"); // 5. 发布消息到交换机 // 5.1 发送10条路由键为"news.music.pop"的消息 for (int i = 0; i < 10; i++) { mymq::BasicProperties bp; bp.set_id(mymq::UUIDHelper::uuid()); // 设置消息唯一ID bp.set_delivery_mode(mymq::DeliveryMode::DURABLE); // 设置消息持久化 bp.set_routing_key("news.music.pop"); // 设置路由键 // 发布消息：交换机名称、消息属性、消息内容 channel->basicPublish("exchange1", &bp, "Hello World-" + std::to_string(i)); } // 5.2 发送一条路由键为"news.music.sport"的消息 mymq::BasicProperties bp; bp.set_id(mymq::UUIDHelper::uuid()); bp.set_delivery_mode(mymq::DeliveryMode::DURABLE); bp.set_routing_key("news.music.sport"); channel->basicPublish("exchange1", &bp, "Hello Bite"); // 5.3 发送一条路由键为"news.sport"的消息 // 注意：修改了同一个bp对象的路由键，会覆盖之前的设置 bp.set_routing_key("news.sport"); channel->basicPublish("exchange1", &bp, "Hello chileme?"); // 6. 关闭信道 conn->closeChannel(channel); return 0; }消费者客户端实现
我们创建一个consume_client.cc，然后把下面这些内容填写进去即可
#include "connection.hpp" #include <iostream> #include <thread> #include <chrono> #include <functional> // 消息消费回调函数 // 参数： // channel: 消息通道指针，用于发送确认消息等操作 // consumer_tag: 消费者标签，标识不同的消费者 // bp: 消息基本属性，包含消息ID、路由键等信息 // body: 消息体内容 void cb(mymq::Channel::ptr &channel, const std::string consumer_tag, const mymq::BasicProperties *bp, const std::string &body) { // 打印消费者信息和接收到的消息内容 std::cout << consumer_tag << "消费了消息：" << body << std::endl; // 发送消息确认，告诉RabbitMQ消息已被成功处理 // 参数：消息唯一ID channel->basicAck(bp->id()); } int main(int argc, char *argv[]) { // 检查命令行参数：需要指定要消费的队列名称 if (argc != 2) { std::cout << "使用方法: ./consume_client <队列名称>\n"; std::cout << "示例: ./consume_client queue1\n"; return -1; } // 1. 创建异步工作线程对象，用于处理网络IO和消息回调 mymq::AsyncWorker::ptr awp = std::make_shared<mymq::AsyncWorker>(); // 2. 建立与RabbitMQ服务器的连接 // 参数：服务器地址、端口号、异步工作线程 mymq::Connection::ptr conn = std::make_shared<mymq::Connection>("127.0.0.1", 8085, awp); // 3. 通过连接创建信道，信道是执行AMQP操作的通道 mymq::Channel::ptr channel = conn->openChannel(); // 4. 声明交换机和队列，并建立绑定关系 // 使用空的属性映射（不设置额外参数） google::protobuf::Map<std::string, std::string> tmp_map; // 4.1 声明一个名为"exchange1"的主题交换机（ExchangeType::TOPIC） // 参数：交换机名称、类型、是否持久化、是否自动删除、额外属性 channel->declareExchange("exchange1", mymq::ExchangeType::TOPIC, true, false, tmp_map); // 4.2 声明一个持久化队列"queue1" // 参数：队列名称、是否持久化、是否排他、是否自动删除、额外属性 channel->declareQueue("queue1", true, false, false, tmp_map); // 4.3 声明一个持久化队列"queue2" channel->declareQueue("queue2", true, false, false, tmp_map); // 4.4 将队列"queue1"绑定到交换机"exchange1"，绑定键为"queue1" // 参数：交换机名称、队列名称、绑定键 channel->queueBind("exchange1", "queue1", "queue1"); // 4.5 将队列"queue2"绑定到交换机"exchange1"，绑定键为"news.music.#" // "#"是通配符，表示匹配多个单词 channel->queueBind("exchange1", "queue2", "news.music.#"); // 5. 使用std::bind创建回调函数适配器 // std::bind将回调函数cb与参数绑定，其中channel作为第一个参数固定传入 // std::placeholders::_1、_2、_3代表回调函数cb的consumer_tag、bp、body参数 auto functor = std::bind(cb, channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3); // 6. 开始消费指定队列的消息 // 参数： // consumer1: 消费者标签，用于标识此消费者 // argv[1]: 队列名称，从命令行参数获取 // false: 是否自动确认消息，false表示需要手动确认（basicAck） // functor: 消息到达时的回调函数 channel->basicConsume("consumer1", argv[1], false, functor); // 7. 保持程序运行，等待消息 // 使用无限循环，每3秒休眠一次，避免CPU占用过高 while(1) { std::this_thread::sleep_for(std::chrono::seconds(3)); } // 8. 关闭信道（这里不会被执行，因为程序一直循环等待消息） conn->closeChannel(channel); return 0; }编译测试
我们可以先进行编译一下，
all: publish_client consume_client publish_client: publish_client.cc ../third/include/muduo/protobuf/codec.cc ../mqcommon/msg.pb.cc ../mqcommon/proto.pb.cc g++ $^ -o $@ -g -I ../third/include -L ../third/lib -lmuduo_net -lmuduo_base -lpthread -lprotobuf -lz -lsqlite3 consume_client: consume_client.cc ../third/include/muduo/protobuf/codec.cc ../mqcommon/msg.pb.cc ../mqcommon/proto.pb.cc g++ $^ -o $@ -g -I ../third/include -L ../third/lib -lmuduo_net -lmuduo_base -lpthread -lprotobuf -lz -lsqlite3 .PHONY: clean clean: rm -rf publish_client consume_client3.2.测试
注意：每次测试之前都需要删除/data目录，这里面存储的是持久化的绑定信息，我们需要将它们删除
没有问题，接下来我们就进入我们的测试环节
首先我们先启动服务器
服务器已经运行起来了
我们先运行消费者客户端
./consume_client queue1换一个终端
./consume_client queue2换一个终端
./publish_client运行之后，我们发现这个queue2的收到了消息
而绑定了queue1的客户端则没有任何反应
