网站虚拟主机1g,wordpress虚线框可序列,wordpress商店模板,软件商店正版下载1. 前言
RabbitMQ是一个流行的开源消息队列系统#xff0c;支持多种消息协议#xff0c;广泛用于构建分布式系统和微服务架构。可以在不同应用程序之间实现异步消息传递。在本文中#xff0c;我们将熟悉如何使用C与RabbitMQ进行消息通信。
2. 准备工作
在 Windows 平台上…
1. 前言
RabbitMQ是一个流行的开源消息队列系统支持多种消息协议广泛用于构建分布式系统和微服务架构。可以在不同应用程序之间实现异步消息传递。在本文中我们将熟悉如何使用C与RabbitMQ进行消息通信。
2. 准备工作
在 Windows 平台上通过 vcpkg 安装 librabbitmq并在 C 中使用该库实现 RabbitMQ 消息的发布和接收。librabbitmq 是官方的 C 客户端库支持与 RabbitMQ 服务器的通信。
2.1 安装 vcpkg
如果还没有安装 vcpkg请按照以下步骤安装 克隆 vcpkg 仓库 git clone https://github.com/microsoft/vcpkg.git进入 vcpkg 目录并运行安装脚本 cd vcpkg
.\bootstrap-vcpkg.bat使用 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 服务器、声明交换机、绑定队列并发送或接收消息。
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);// 声明交换机类型为 directamqp_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 消息发送和接收的完整教程。
