news 2026/5/8 15:12:18

【消息队列项目】客户端搭建与测试

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
【消息队列项目】客户端搭建与测试

目录

一.广播交换模式下的测试

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_client

1.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_client

3.2.测试

注意：每次测试之前都需要删除/data目录，这里面存储的是持久化的绑定信息，我们需要将它们删除

没有问题，接下来我们就进入我们的测试环节

首先我们先启动服务器

服务器已经运行起来了

我们先运行消费者客户端

./consume_client queue1

换一个终端

./consume_client queue2

换一个终端

./publish_client

运行之后，我们发现这个queue2的收到了消息

而绑定了queue1的客户端则没有任何反应

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/5/3 17:33:30

7678678

687678687

作者头像 李华
网站建设 2026/5/4 9:19:13

Jupyter Notebook保存为Markdown格式：方便技术文档输出

Jupyter Notebook 保存为 Markdown&#xff1a;让实验成果轻松转化为技术文档 在深度学习项目中&#xff0c;我们常常会经历这样的场景&#xff1a;花了几天时间在 Jupyter Notebook 里调通模型、画出关键图表、写下分析逻辑&#xff0c;最后却卡在“怎么把这一切讲清楚”这一…

作者头像 李华
网站建设 2026/5/8 1:24:45

Conda环境备份迁移：复制现有PyTorch配置到新机器

Conda环境备份迁移&#xff1a;复制现有PyTorch配置到新机器 在深度学习项目中&#xff0c;最让人头疼的往往不是模型调参&#xff0c;而是“在我电脑上明明能跑”的环境问题。一个团队里五个人装环境&#xff0c;最后可能配出三种不同的行为结果——有人CUDA不识别&#xff0c…

作者头像 李华
网站建设 2026/4/27 15:03:23

Conda列出已安装包：筛选出与PyTorch相关的库

Conda筛选PyTorch相关包&#xff1a;高效验证深度学习环境完整性的实践指南 在深度学习项目中&#xff0c;最令人沮丧的场景之一莫过于代码写完准备训练时&#xff0c;却突然报出 ModuleNotFoundError: No module named torch。更糟的是&#xff0c;在远程服务器或团队共享环境…

作者头像 李华
网站建设 2026/4/23 10:34:11

Conda环境删除恢复：误删后如何找回PyTorch配置

Conda环境删除恢复&#xff1a;误删后如何找回PyTorch配置 在深度学习项目开发中&#xff0c;一个稳定的运行环境往往比代码本身更“脆弱”。你可能花了一整天调试好 PyTorch CUDA 的版本组合&#xff0c;结果一条 conda remove -n pytorch_env --all 命令误执行&#xff0c;…

作者头像 李华
网站建设 2026/5/5 21:39:21

Conda环境变量设置：指定CUDA_VISIBLE_DEVICES控制GPU使用

Conda环境变量设置&#xff1a;指定CUDA_VISIBLE_DEVICES控制GPU使用 在现代深度学习开发中&#xff0c;我们经常面对这样一个现实&#xff1a;服务器上插着四块A100显卡&#xff0c;但你只想用其中一块跑实验&#xff0c;而同事正占用另一张卡训练大模型。如果程序一启动就抢占…

作者头像 李华