news 2026/4/16 12:01:35

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

作者头像

张小明

前端开发工程师

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/4/16 11:03:26

SSH隧道转发TensorBoard端口：本地可视化远程训练指标

SSH隧道转发TensorBoard端口&#xff1a;本地可视化远程训练指标 在深度学习的实际开发中&#xff0c;一个再熟悉不过的场景是&#xff1a;你在办公室或家里的笔记本上敲代码&#xff0c;而真正的“算力战场”却远在数据中心的一台搭载A100的服务器上。模型正在那里安静地训练…

作者头像 李华
网站建设 2026/4/16 11:00:53

GitHub Release发布正式版PyTorch模型权重文件

PyTorch-CUDA-v2.8 镜像&#xff1a;从模型权重到可复现环境的一体化交付 在深度学习项目中&#xff0c;你是否经历过这样的场景&#xff1f; 同事发来一个“训练好的模型权重”&#xff0c;你兴冲冲地下载下来准备推理测试&#xff0c;结果运行第一行代码就报错&#xff1a;CU…

作者头像 李华
网站建设 2026/4/11 7:59:13

清华镜像站同步频率解析：确保PyTorch包版本最新

清华镜像站同步频率解析&#xff1a;确保PyTorch包版本最新 在人工智能研发一线&#xff0c;你是否曾经历过这样的场景&#xff1a;深夜调试模型&#xff0c;pip install torch 却卡在 30% 进度条上动弹不得&#xff1f;或者团队成员因 PyTorch 版本不一致导致实验结果无法复现…

作者头像 李华
网站建设 2026/4/15 14:46:29

PyTorch-CUDA-v2.8镜像资源调度优化方向探讨

PyTorch-CUDA-v2.8镜像资源调度优化方向探讨 在当前深度学习项目日益复杂、训练任务频繁迭代的背景下&#xff0c;一个稳定、高效且可复用的运行时环境已成为研发流程中的关键基础设施。尤其是在多团队协作、GPU集群共享或持续集成&#xff08;CI/CD&#xff09;场景下&#xf…

作者头像 李华
网站建设 2026/4/15 8:12:20

DTD 元素

DTD 元素 概述 DTD（Document Type Definition，文档类型定义）是XML文档的骨架，它定义了XML文档的结构、元素、属性和它们的约束关系。DTD是一种简单的XML文档声明，用于描述XML文档的结构。它主要被用于定义XML文档的类型，确保XML文档的合法性。 DTD的基本结构 一个DTD…

作者头像 李华
网站建设 2026/4/14 9:20:52

YOLOv5++改进版在PyTorch-CUDA-v2.8上的适配尝试

YOLOv5 在 PyTorch-CUDA-v2.8 环境下的适配实践 在当前 AI 工程化加速落地的背景下&#xff0c;一个常见的现实挑战浮出水面&#xff1a;如何让社区中那些“纸面性能亮眼”的改进模型&#xff0c;真正跑起来&#xff1f;以 YOLOv5 为例&#xff0c;这个由开发者社区自发演进的高…

作者头像 李华