项目第十二弹：功能联调
- 一、发布订阅功能测试
- 1.生产者
- 2.消费者
- 3.演示
- 4.持久化信息查看
- 1.消息
- 2.SQLite3数据库
- 二、持久化恢复测试
- 1.代码
- 2.gc
- 3.演示
- 三、虚拟机和信道隔离测试
- 1.责任划分
- 2.如何测试
- 3.生产者
- 4.消费者
- 5.演示
一、发布订阅功能测试
我们直接上TOPIC交换机类型了，主题交换没事的话，那么DIRECT和FANOUT肯定也没事
路由匹配模块我们也进行过单元测试，不可能有任何问题
服务器代码：
#include "broker.hpp"int main()
{ns_mq::Server server(8888);server.start();return 0;
}
1.生产者
#include "connection.hpp"
using namespace ns_mq;int main()
{AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();// 1. 创建连接和信道Connection::ptr conn = std::make_shared<Connection>("127.0.0.1", 8888, worker);Channel::ptr cp = conn->OpenChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host1", "queue1", true, false, false, {});cp->declareMsgQueue("host1", "queue2", true, false, false, {});cp->bind("host1", "exchange1", "queue1", "news.sport.#");cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 发送10条消息BasicProperities bp;bp.set_mode(DURABLE);bp.set_routing_key("news.sport.basketball");for (int i = 0; i < 10; i++){bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "Hello -" + std::to_string(i));}bp.set_routing_key("news.singer.zhangsan");bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "singer zhangsan");bp.set_routing_key("news.sportsman.zhangsan");bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "sportsman zhangsan");// 4. 关闭信道conn->CloseChannel(cp);return 0;
}
2.消费者
#include "connection.hpp"
using namespace ns_mq;// 因为要拿到信道才能进行确认,所以这里需要把Channel::ptr bind过来
void Callback(const Channel::ptr &cp, const std::string &consumer_tag, const BasicProperities *bp, const std::string &body)
{// 1. 消费消息std::string id;if (bp != nullptr){id = bp->msg_id();}std::cout << consumer_tag << " 消费了消息: " << body << ", 消息ID: " << id << "\n";// 2. 确认消息if (bp != nullptr)cp->BasicAck(id);
}// consumer_tag queue_name
int main(int argc, char *argv[])
{if (argc != 3){std::cout << "Usage: " << argv[0] << " consumer_tag queue_name\n";return 1;}AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();// 1. 创建连接和信道Connection::ptr conn = std::make_shared<Connection>("127.0.0.1", 8888, worker);Channel::ptr cp = conn->OpenChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host1", "queue1", true, false, false, {});cp->declareMsgQueue("host1", "queue2", true, false, false, {});cp->bind("host1", "exchange1", "queue1", "news.sport.#");cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 创建消费者cp->BasicConsume("host1", argv[1], argv[2],std::bind(Callback, cp, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), false);// 4. 等待消息while (true){std::this_thread::sleep_for(std::chrono::seconds(1000));}// 5. 关闭信道conn->CloseChannel(cp);return 0;
}
3.演示
4.持久化信息查看
1.消息
2.SQLite3数据库
二、持久化恢复测试
1.代码
生产者：
#include "connection.hpp"
using namespace ns_mq;int main()
{AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();// 1. 创建连接和信道Connection::ptr conn = std::make_shared<Connection>("127.0.0.1", 8888, worker);Channel::ptr cp = conn->OpenChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定// cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");// cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});// cp->declareMsgQueue("host1", "queue1", true, false, false, {});// cp->declareMsgQueue("host1", "queue2", true, false, false, {});// cp->bind("host1", "exchange1", "queue1", "news.sport.#");// cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 发送10条消息BasicProperities bp;bp.set_mode(DURABLE);bp.set_routing_key("news.sport.basketball");for (int i = 0; i < 10; i++){bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "Hello -" + std::to_string(i));}bp.set_routing_key("news.singer.zhangsan");bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "singer zhangsan");bp.set_routing_key("news.sportsman.zhangsan");bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "sportsman zhangsan");// 4. 关闭信道conn->CloseChannel(cp);return 0;
}
消费者：
#include "connection.hpp"
using namespace ns_mq;// 因为要拿到信道才能进行确认,所以这里需要把Channel::ptr bind过来
void Callback(const Channel::ptr &cp, const std::string &consumer_tag, const BasicProperities *bp, const std::string &body)
{// 1. 消费消息std::string id;if (bp != nullptr){id = bp->msg_id();}std::cout << consumer_tag << " 消费了消息: " << body << ", 消息ID: " << id << "\n";// 2. 确认消息if (bp != nullptr)cp->BasicAck(id);
}// consumer_tag queue_name
int main(int argc, char *argv[])
{if (argc != 3){std::cout << "Usage: " << argv[0] << " consumer_tag queue_name\n";return 1;}AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();// 1. 创建连接和信道Connection::ptr conn = std::make_shared<Connection>("127.0.0.1", 8888, worker);Channel::ptr cp = conn->OpenChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定// cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");// cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});// cp->declareMsgQueue("host1", "queue1", true, false, false, {});// cp->declareMsgQueue("host1", "queue2", true, false, false, {});// cp->bind("host1", "exchange1", "queue1", "news.sport.#");// cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 创建消费者cp->BasicConsume("host1", argv[1], argv[2],std::bind(Callback, cp, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), false);// 4. 等待消息while (true){std::this_thread::sleep_for(std::chrono::seconds(1000));}// 5. 关闭信道conn->CloseChannel(cp);return 0;
}
2.gc
3.演示
不再次创建任何交换机，队列，绑定信息，看是否能够继续发布消息
完美：
三、虚拟机和信道隔离测试
下面我们开多线程测一测信道和虚拟机：
信道实现了访问上的隔离，Vhost实现了资源上的隔离
两者配合使用，我们便能够在一个进程当中通过多线程完美的实现RabbitMQ的用户隔离
1.责任划分
RabbitMQ当中，多线程使用不同的信道，如果两个线程都对同一个队列进行操作，比如，一个线程向该队列绑定消费者，另一个线程删除了该队列。按道理来说，即使出现了问题，也不能怪RabbitMQ的信道，不能怪RabbitMQ。
因为你用户使用多线程没有好好使用，在你的代码当中，那个队列已经同时被两个线程访问了，而且某个线程还对该队列进行了写操作，因此这个队列对两个线程来说就已经是临界资源了，你访问这个临界资源不加锁，这是你用户的问题，不是RabbitMQ的问题
其实就像信号量一样，约定好哪个线程负责哪些资源，对资源/任务进行无冲突的划分即可这是用户使用多线程访问RabbitMQ必须要做的，这也是多线程并发编程最基本的原则
2.如何测试
虚拟机跟虚拟机之间是具有隔离性的，不同虚拟机之间可以创建同名交换机，队列，绑定信息，可以具有同名消费者，同msg_id的消息等等…
但是在同一虚拟机当中，不能具有同名交换机…等等
因此我们就让不同虚拟机有同名的交换机，队列，绑定信息…
就是这么测，测的就是虚拟机和信道的隔离性
3.生产者
#include "connection.hpp"
using namespace ns_mq;
#include <thread>
#include <vector>
using namespace std;// host1
void publisher1(const Connection::ptr &conn, const std::string &thread_name)
{// 1. 创建信道Channel::ptr cp = conn->OpenChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host1", "queue1", true, false, false, {});//cp->declareMsgQueue("host1", "queue2", true, false, false, {});cp->bind("host1", "exchange1", "queue1", "news.sport.#");//cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 发送10条消息BasicProperities bp;bp.set_mode(DURABLE);bp.set_routing_key("news.sport.basketball");for (int i = 0; i < 5; i++){bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "Hello -" + std::to_string(i));}// 4. 关闭信道conn->CloseChannel(cp);
}// host2
void publisher2(const Connection::ptr &conn, const std::string &thread_name)
{// 1. 创建信道Channel::ptr cp = conn->OpenChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host2", "./host2/resource.db", "./host2/message");cp->declareExchange("host2", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host2", "queue1", true, false, false, {});cp->declareMsgQueue("host2", "queue2", true, false, false, {});cp->bind("host2", "exchange1", "queue1", "news.sport.#");cp->bind("host2", "exchange1", "queue2", "news.*.zhangsan");// 3. 发送10条消息BasicProperities bp;bp.set_mode(DURABLE);bp.set_routing_key("news.sport.basketball");for (int i = 0; i < 5; i++){bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host2", "exchange1", &bp, "Hello -" + std::to_string(i));}// 4. 关闭信道conn->CloseChannel(cp);
}int main()
{AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();Connection::ptr myconn = std::make_shared<Connection>("127.0.0.1", 8888, worker);vector<thread> thread_v;thread_v.push_back(thread(publisher1, myconn, "thread1"));thread_v.push_back(thread(publisher2, myconn, "thread2"));for(auto& t:thread_v)t.join();return 0;
}
4.消费者
#include "connection.hpp"
using namespace ns_mq;
#include <thread>
#include <vector>
using namespace std;// 因为要拿到信道才能进行确认,所以这里需要把Channel::ptr bind过来
void Callback(const Channel::ptr &cp, const std::string &consumer_tag, const BasicProperities *bp, const std::string &body)
{// 1. 消费消息std::string id;if (bp != nullptr){id = bp->msg_id();}std::cout << consumer_tag << " 消费了消息: " << body << ", 消息ID: " << id << "\n";// 2. 确认消息if (bp != nullptr)std::cout << cp->BasicAck(id) << "\n";
}void consumer1(const Connection::ptr &conn, const std::string &thread_name)
{Channel::ptr cp = conn->OpenChannel();std::cout << "consumer1: 信道ID：" << cp->cid() << "\n";// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host1", "queue1", true, false, false, {});//cp->declareMsgQueue("host1", "queue2", true, false, false, {});cp->bind("host1", "exchange1", "queue1", "news.sport.#");//cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 创建消费者cp->BasicConsume("host1", "consumer1", "queue1",std::bind(Callback, cp, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), false);// 4. 等待消息while (true){std::this_thread::sleep_for(std::chrono::seconds(1000));}// 5. 关闭信道conn->CloseChannel(cp);
}void consumer2(const Connection::ptr &conn, const std::string &thread_name)
{Channel::ptr cp = conn->OpenChannel();std::cout << "consumer2: 信道ID：" << cp->cid() << "\n";// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host2", "./host2/resource.db", "./host2/message");cp->declareExchange("host2", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host2", "queue1", true, false, false, {});cp->declareMsgQueue("host2", "queue2", true, false, false, {});cp->bind("host2", "exchange1", "queue1", "news.sport.#");cp->bind("host2", "exchange1", "queue2", "news.*.zhangsan");// 3. 创建消费者cp->BasicConsume("host2", "consumer2", "queue1",std::bind(Callback, cp, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), false);// 4. 等待消息while (true){std::this_thread::sleep_for(std::chrono::seconds(1000));}// 5. 关闭信道conn->CloseChannel(cp);
}// consumer_tag queue_name
int main()
{AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();// 1. 创建连接和信道Connection::ptr conn = std::make_shared<Connection>("127.0.0.1", 8888, worker);vector<thread> thread_v;thread_v.push_back(thread(consumer1, conn, "thread1"));thread_v.push_back(thread(consumer2, conn, "thread2"));for (auto &t : thread_v)t.join();return 0;
}
5.演示
可见，正是因为RabbitMQ有信道和虚拟机，所以使用体验大幅度上升，功能更加灵活，他们两个是RabbitMQ非常好的设计
信道：
- 性能优化（网络层面：连接细分为信道，从而复用TCP连接）
- 细粒度控制（服务层面：通过信道来进行服务的提供，【会话管理】更加精细化）
- 资源隔离（访问句柄的资源隔离）
虚拟机：
- 多租户支持（实际的资源隔离）
- 配置灵活（用户认证，权限管理…）
以上就是项目第十二弹：功能联调的全部内容
