C++开发基础之使用librabbitmq库实现RabbitMQ消息队列通信

news/2024/11/14 0:41:51/

在这里插入图片描述

1. 前言

RabbitMQ是一个流行的开源消息队列系统,支持多种消息协议,广泛用于构建分布式系统和微服务架构。可以在不同应用程序之间实现异步消息传递。在本文中,我们将熟悉如何使用C++与RabbitMQ进行消息通信。

2. 准备工作

在 Windows 平台上通过 vcpkg 安装 librabbitmq,并在 C++ 中使用该库实现 RabbitMQ 消息的发布和接收。librabbitmq 是官方的 C 客户端库,支持与 RabbitMQ 服务器的通信。在这里插入图片描述

2.1 安装 vcpkg

如果还没有安装 vcpkg,请按照以下步骤安装:

  1. 克隆 vcpkg 仓库:

    git clone https://github.com/microsoft/vcpkg.git
    
  2. 进入 vcpkg 目录并运行安装脚本:

    cd vcpkg
    .\bootstrap-vcpkg.bat
    
  3. 使用 vcpkg 安装 RabbitMQ C 库(librabbitmq):

    vcpkg install librabbitmq
    

在这里插入图片描述

安装完成后,将 vcpkg 集成到项目中:

vcpkg integrate install

这样,librabbitmq 库会被自动链接到 Visual Studio 中的 C++ 项目。
在这里插入图片描述

2.2 配置 Visual Studio 项目

在 Visual Studio 中创建新的 C++ 项目,确保项目中包含了 vcpkg 的安装路径。vcpkg 会自动配置项目,使其能够找到并链接到 librabbitmq 库。并且链接器附加依赖项添加rabbitmq.4.lib便于程序查找rabbitmq.4.dll库引用。
在这里插入图片描述

3. 实现消息发送和接收程序

我们将编写两个程序,一个用于发送消息,一个用于接收消息。这些程序将演示如何使用 librabbitmq 库连接到 RabbitMQ 服务器、声明交换机、绑定队列并发送或接收消息。

rabbitmq_Server_41">3.1 启动rabbitmq Server

在这里插入图片描述

3.2 发送消息的程序(Producer)

以下是消息发送者的完整代码,它会循环发送多条消息到指定的 RabbitMQ 交换机和队列。

#include <iostream>
#include <string>
#include <amqp.h>
#include <amqp_tcp_socket.h>
#include <thread>
#include <chrono>// 用于处理 AMQP 错误并输出错误信息
void die_on_error(amqp_rpc_reply_t x, const char* context) {if (x.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Error in " << context << ": "<< amqp_error_string2(x.library_error) << std::endl;exit(1);}
}int main() {const std::string hostname = "localhost";  // RabbitMQ 服务器地址const int port = 5672;  // RabbitMQ 默认端口const std::string exchange = "example_exchange";  // 交换机名称const std::string routing_key = "example_key";  // 路由键,用于绑定队列// 初始化连接amqp_connection_state_t conn = amqp_new_connection();amqp_socket_t* socket = amqp_tcp_socket_new(conn);if (!socket) {std::cerr << "Creating TCP socket failed" << std::endl;return 1;}// 打开 TCP 连接int status = amqp_socket_open(socket, hostname.c_str(), port);if (status) {std::cerr << "Opening TCP socket failed" << std::endl;return 1;}// 登录 RabbitMQdie_on_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in");amqp_channel_open(conn, 1);  // 打开信道die_on_error(amqp_get_rpc_reply(conn), "Opening channel");// 声明交换机(类型为 direct)amqp_exchange_declare(conn, 1, amqp_cstring_bytes(exchange.c_str()), amqp_cstring_bytes("direct"),0, 0, 0, 0, amqp_empty_table);die_on_error(amqp_get_rpc_reply(conn), "Declaring exchange");// 循环发送多条消息for (int i = 1; i <= 1000; ++i) {  // 发送 1000 条消息std::string message = "Hello, RabbitMQ! Message number: " + std::to_string(i);amqp_bytes_t message_bytes = amqp_cstring_bytes(message.c_str());// 设置消息属性amqp_basic_properties_t props;props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;props.content_type = amqp_cstring_bytes("text/plain");props.delivery_mode = 2;  // 持久化模式// 发送消息到交换机int result = amqp_basic_publish(conn, 1, amqp_cstring_bytes(exchange.c_str()), amqp_cstring_bytes(routing_key.c_str()),0, 0, &props, message_bytes);if (result < 0) {std::cerr << "Error publishing message " << i << std::endl;} else {std::cout << "Message " << i << " published: " << message << std::endl;}// 每次发送后等待 1 秒std::this_thread::sleep_for(std::chrono::seconds(1));}// 清理连接amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);amqp_connection_close(conn, AMQP_REPLY_SUCCESS);amqp_destroy_connection(conn);return 0;
}

执行结果
在这里插入图片描述

3.3 接收消息的程序(Consumer)

以下是接收消息的完整代码,使用 amqp_consume_message 接收并打印消息内容。

#include <iostream>
#include <string>
#include <amqp.h>
#include <amqp_tcp_socket.h>// 错误处理函数,用于输出错误信息
void die_on_error(amqp_rpc_reply_t x, const char* context) {if (x.reply_type != AMQP_RESPONSE_NORMAL) {std::cerr << "Error in " << context << ": "<< amqp_error_string2(x.library_error) << std::endl;exit(1);}
}int main() {const std::string hostname = "localhost";  // RabbitMQ 服务器地址const int port = 5672;  // 端口const std::string queue = "example_queue";  // 队列名称const std::string exchange = "example_exchange";  // 交换机名称const std::string routing_key = "example_key";  // 路由键// 初始化连接amqp_connection_state_t conn = amqp_new_connection();amqp_socket_t* socket = amqp_tcp_socket_new(conn);if (!socket) {std::cerr << "Creating TCP socket failed" << std::endl;return 1;}// 打开 TCP 连接int status = amqp_socket_open(socket, hostname.c_str(), port);if (status) {std::cerr << "Opening TCP socket failed" << std::endl;return 1;}// 登录 RabbitMQdie_on_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in");amqp_channel_open(conn, 1);die_on_error(amqp_get_rpc_reply(conn), "Opening channel");// 声明交换机和队列,并绑定队列到交换机amqp_exchange_declare(conn, 1, amqp_cstring_bytes(exchange.c_str()), amqp_cstring_bytes("direct"),0, 0, 0, 0, amqp_empty_table);die_on_error(amqp_get_rpc_reply(conn), "Declaring exchange");amqp_queue_declare_ok_t* q = amqp_queue_declare(conn, 1, amqp_cstring_bytes(queue.c_str()), 0, 0, 0, 1, amqp_empty_table);die_on_error(amqp_get_rpc_reply(conn), "Declaring queue");amqp_queue_bind(conn, 1, amqp_cstring_bytes(queue.c_str()), amqp_cstring_bytes(exchange.c_str()), amqp_cstring_bytes(routing_key.c_str()), amqp_empty_table);die_on_error(amqp_get_rpc_reply(conn), "Binding queue");// 开始消费消息amqp_basic_consume(conn, 1, amqp_cstring_bytes(queue.c_str()), amqp_empty_bytes, 0, 1, 0, amqp_empty_table);die_on_error(amqp_get_rpc_reply(conn), "Consuming");while (true) {amqp_rpc_reply_t res;amqp_envelope_t envelope;// 释放资源amqp_maybe_release_buffers(conn);res = amqp_consume_message(conn, &envelope, NULL, 0);// 检查并打印接收到的消息if (res.reply_type == AMQP_RESPONSE_NORMAL) {std::cout << "Received: " << std::string((char*)envelope.message.body.bytes, envelope.message.body.len) << std::endl;amqp_destroy_envelope(&envelope);} else {std::cerr << "Error consuming message" << std::endl;break;}}

执行结果
在这里插入图片描述
可以查看RabbitMQ的webUI,了解消息的投递和消费情况
在这里插入图片描述

4. 总结

我们已完成了在 Windows 平台上通过 vcpkg 安装 librabbitmq 并用 C++ 实现 RabbitMQ 消息发送和接收的完整教程。


http://www.ppmy.cn/news/1546784.html

相关文章

计算机新手练级攻略——如何搜索问题

目录 计算机学生新手练级攻略——如何搜索问题1.明确搜索意图2.使用精确关键词3.使用专业引擎搜索4.利用好技术社区1. Stack Overflow2. GitHub3. IEEE Xplore4. DBLP 5.使用代码搜索工具1. GitHub 代码搜索2. Stack Overflow 代码搜索3. Papers with Code4. IEEE Xplore 6.查阅…

【STM32】GPIO口以及EXTI外部中断

在STM32微控制器中&#xff0c;GPIO&#xff08;通用输入输出&#xff09;口和EXTI&#xff08;外部中断&#xff09;是实现用户交互和事件响应的重要功能。本文将详细介绍STM32的GPIO口和EXTI外部中断的配置和使用方法&#xff0c;以及如何通过代码实现这些功能。 GPIO口简介…

创新体验触手可及 紫光展锐携手影目科技推出AI眼镜开放平台

近日&#xff0c;紫光展锐携手影目科技共同发布了搭载展锐W517芯片的影目X系列AI眼镜开放平台。这一产品的推出标志着双方在智能穿戴领域的深度协作&#xff0c;将紫光展锐的领先芯片技术与影目的产品创新相融合&#xff0c;合力打造全球智能眼镜市场的标杆产品。这一战略布局不…

vue3中的组件通信方式有哪些?

在vue3里&#xff0c;组件是一个非常重要的概念&#xff0c;项目中各个组件间的通信也是一个非常常见的需求&#xff0c;接下来我将为大家展示vue3组件有哪几种常见的通信方式。 一、props 适用场景&#xff1a;父子组件之间的通信 父传子&#xff1a; 父组件在子组件的标签…

HarmonyOS入门简介

&#x1f341; 作者&#xff1a;知识浅谈&#xff0c;CSDN签约讲师&博客专家&#xff0c;华为云云享专家&#xff0c;阿里云专家博主&#xff0c;InfoQ签约作者 &#x1f4cc; 擅长领域&#xff1a;全栈工程师、爬虫、ACM算法&#xff0c;大数据&#xff0c;深度学习 &…

软考系统架构设计师论文:论边缘计算及其应用

试题一 论边缘计算及其应用 边缘计算是在靠近物或数据源头的网络边缘侧,融合网络、计算、存储、应用核心能力的 分布式开放平台(架构),就近提供边缘智能服务。边缘计算与云计算各有所长,云计算擅长全局性、非实时、长周期的大数据处理与分析,能够在长周期维护、业务决策…

sudo docker ps才能查看,docker ps不能查看问题

出现 permission denied while trying to connect to the Docker daemon socket 的错误&#xff0c;通常是因为当前用户没有权限访问 Docker 的 Unix 套接字 /var/run/docker.sock。在 Linux 系统中&#xff0c;这个套接字默认只能由 root 用户或 docker 组的成员访问。 要解决…

深入了解区块链:Web3的基础架构与发展

区块链技术是Web3生态系统的核心&#xff0c;它正推动互联网走向去中心化。Web3不仅代表着技术的创新&#xff0c;也挑战着现有互联网结构和用户数据的控制方式。区块链作为Web3的基础&#xff0c;正逐步改变我们对网络交互和数据管理的认知。本文将介绍区块链的基本原理及其在…