[C++][第三方库][RabbitMq]详细讲解

news/2024/10/21 10:08:09/

目录

  • 1.介绍
  • 2.安装
    • 1.RabbitMq
    • 2.客户端库
  • 3.AMQP-CPP 简单使用
    • 1.介绍
    • 2.使用
  • 4.类与接口
    • 1.Channel
    • 2.ev
  • 5.使用
    • 1.publish.cc
    • 2.consume.cc
    • 3.makefile

1.介绍

  • RabbitMQ：消息队列组件，实现两个客户端主机之间消息传输的功能(发布&订阅)
  • 核心概念：交换机、队列、绑定、消息
  • 交换机类型：
    • 广播交换：当交换机收到消息，则将消息发布到所有绑定的队列中
    • 直接交换：根据消息中的bkey与绑定的rkey对比，一致则放入队列
    • 主题交换：使用bkey与绑定的rkey进行规则匹配，成功则放入队列

2.安装

1.RabbitMq

  • 安装：sudo apt install rabbitmq-server
  • 简单使用：
    # 安装完成的时候默认有个用户guest，但是权限不够，要创建一个administrator用户，才可以做为远程登录和发表订阅消息#添加用户 
sudo rabbitmqctl add_user root <PASSWORD>#设置用户tag 
sudo rabbitmqctl set_user_tags root administrator #设置用户权限 
sudo rabbitmqctl set_permissions -p / root "." "." ".*" # RabbitMQ自带了web管理界面, 执行下面命令开启, 默认端口15672
sudo rabbitmq-plugins enable rabbitmq_management

2.客户端库

  • C语言库
  • C++库
    sudo apt install libev-dev #libev 网络库组件
git clone https://github.com/CopernicaMarketingSoftware/AMQP-CPP.git
cd AMQP-CPP/
make
make install
  • 如果安装时出现以下报错，则表示ssl版本出现问题
    /usr/include/openssl/macros.h:147:4: error: #error 
"OPENSSL_API_COMPAT expresses an impossible API compatibility 
level" 147 | #  error "OPENSSL_API_COMPAT expresses an impossible API 
compatibility level" |    ^~~~~ 
In file included from /usr/include/openssl/ssl.h:18, from linux_tcp/openssl.h:20, from linux_tcp/openssl.cpp:12: 
/usr/include/openssl/bio.h:687:1: error: expected constructor, 
destructor, or type conversion before ‘DEPRECATEDIN_1_1_0’ 687 | DEPRECATEDIN_1_1_0(int BIO_get_port(const char *str, 
unsigned short *port_ptr))
  • 解决方案：卸载当前的ssl库，重新进行修复安装
    dpkg -l | grep ssl
sudo dpkg -P --force-all libevent-openssl-2.1-7
sudo dpkg -P --force-all openssl
sudo dpkg -P --force-all libssl-dev
sudo apt --fix-broken install

3.AMQP-CPP 简单使用

1.介绍

  • AMQP-CPP是用于与RabbitMq消息中间件通信的C++库
    • 它能解析从RabbitMq服务发送来的数据，也可以生成发向RabbitMq的数据包
    • AMQP-CPP库不会向RabbitMq建立网络连接，所有的网络IO由用户完成
  • AMQP-CPP提供了可选的网络层接口，它预定义了TCP模块，用户就不用自己实现网络IO，
    • 也可以选择libevent、libev、libuv、asio等异步通信组件， 需要手动安装对应的组件
  • AMQP-CPP完全异步，没有阻塞式的系统调用，不使用线程就能够应用在高性能应用中
  • 注意：它需要C++17的支持

2.使用

  • AMQP-CPP的使用有两种模式：
    • 使用默认的TCP模块进行网络通信
    • 使用扩展的libevent、libev、libuv、asio异步通信组件进行通信
  • 此处以libev为例，不需要自己实现monitor函数，可以直接使用AMQP::LibEvHandler

4.类与接口

1.Channel

  • channel是一个虚拟连接，一个连接上可以建立多个通道
    • 并且所有的RabbitMq指令都是通过channel传输
      • 所以连接建立后的第一步，就是建立channel
    • 因为所有操作是异步的，所以在channel上执行指令的返回值并不能作为操作执行结果
      • 实际上它返回的是Deferred类，可以使用它安装处理函数
namespace AMQP 
{ /** *  Generic callbacks that are used by many deferred objects */ using SuccessCallback = std::function<void()>; using ErrorCallback = std::function<void(const char *message)>; using FinalizeCallback = std::function<void()>;/** *  Declaring and deleting a queue */ using QueueCallback = std::function<void(const std::string &name, uint32_t messagecount, uint32_t consumercount)>;using DeleteCallback = std::function<void(uint32_t deletedmessages)>; using MessageCallback = std::function<void(const Message &message, uint64_t deliveryTag, bool redelivered)>; // 当使用发布者确认时，当服务器确认消息已被接收和处理时，将调用AckCallback using AckCallback = std::function<void(uint64_t deliveryTag, bool multiple)>;// 使用确认包裹通道时，当消息被ack/nacked时，会调用这些回调 using PublishAckCallback = std::function<void()>; using PublishNackCallback = std::function<void()>; using PublishLostCallback = std::function<void()>; // 信道类class Channel { Channel(Connection *connection); bool connected();/** *声明交换机 *如果提供了一个空名称，则服务器将分配一个名称。 *以下flags可用于交换机： * *-durable     持久化，重启后交换机依然有效 *-autodelete  删除所有连接的队列后，自动删除交换 *-passive     仅被动检查交换机是否存在 *-internal    创建内部交换 * *@param name    交换机的名称 *@param-type    交换类型 enum ExchangeType { fanout,  广播交换，绑定的队列都能拿到消息 direct,  直接交换，只将消息交给routingkey一致的队列 topic,   主题交换，将消息交给符合bindingkey规则的队列 headers, consistent_hash, message_deduplication }; *@param flags    交换机标志 *@param arguments其他参数 * *此函数返回一个延迟处理程序。可以安装回调 using onSuccess(), onError() and onFinalize() methods. */ Deferred &declareExchange(const std::string_view &name,ExchangeType type,int flags,const Table &arguments);/** *声明队列 *如果不提供名称，服务器将分配一个名称。 *flags可以是以下值的组合： * *-durable 持久队列在代理重新启动后仍然有效 *-autodelete 当所有连接的使用者都离开时，自动删除队列 *-passive 仅被动检查队列是否存在 *-exclusive 队列仅存在于此连接，并且在连接断开时自动删除 * *@param name        队列的名称 *@param flags       标志组合 *@param arguments  可选参数 * *此函数返回一个延迟处理程序。可以安装回调 *使用onSuccess（）、onError（）和onFinalize（）方法。 * Deferred &onError(const char *message) * *可以安装的onSuccess（）回调应该具有以下签名： void myCallback(const std::string &name,  uint32_t messageCount,  uint32_t consumerCount); 例如： channel.declareQueue("myqueue").onSuccess( [](const std::string &name,  uint32_t messageCount, uint32_t consumerCount) { std::cout << "Queue '" << name << "' "; std::cout << "has been declared with "; std::cout << messageCount; std::cout << " messages and "; std::cout << consumerCount; std::cout << " consumers" << std::endl; *  }); */ DeferredQueue &declareQueue(const std::string_view &name,int flags,const Table &arguments);/** *将队列绑定到交换机 * *@param exchange     源交换机 *@param queue        目标队列 *@param routingkey   路由密钥 *@param arguments    其他绑定参数 * *此函数返回一个延迟处理程序。可以安装回调 *使用onSuccess（）、onError（）和onFinalize（）方法。 */ Deferred &bindQueue(const std::string_view &exchange,const std::string_view &queue,const std::string_view &routingkey,const Table &arguments);/** *将消息发布到exchange*您必须提供交换机的名称和路由密钥。 然后，RabbitMQ将尝试将消息发送到一个或多个队列。 使用可选的flags参数，可以指定如果消息无法路由到队列时应该发生的情况。默认情况下，不可更改的消息将被静默地丢弃。 *  *如果设置了'mandatory'或'immediate'标志， 则无法处理的消息将返回到应用程序。 在开始发布之前，请确保您已经调用了recall（）-方法， 并设置了所有适当的处理程序来处理这些返回的消息。 *  *可以提供以下flags： *  *-mandatory 如果设置，服务器将返回未发送到队列的消息 *-immediate 如果设置，服务器将返回无法立即转发给使用者的消息。 *@param exchange要发布到的交易所 *@param routingkey路由密钥 *@param envelope要发送的完整信封 *@param message要发送的消息 *@param size消息的大小 *@param flags可选标志 */ bool publish(const std::string_view &exchange,const std::string_view &routingKey,const std::string &message,int flags = 0);/** *告诉RabbitMQ服务器已准备好使用消息-也就是 订阅队列消息 * *调用此方法后，RabbitMQ开始向客户端应用程序传递消息。 consumer tag是一个字符串标识符， 如果您以后想通过channel:：cancel（）调用停止它， 可以使用它来标识使用者。 *如果您没有指定使用者tag，服务器将为您分配一个。 * *支持以下flags： * *-nolocal    如果设置了，则不会同时消耗在此通道上发布的消息 *-noack      如果设置了，则不必对已消费的消息进行确认 *-exclusive  请求独占访问，只有此使用者可以访问队列 * *@param queue    您要使用的队列 *@param tag      将与此消费操作关联的消费者标记 *@param flags    其他标记 *@param arguments其他参数 * *此函数返回一个延迟处理程序。 可以使用onSuccess（）、onError（）和onFinalize（）方法安装回调可以安装的onSuccess（）回调应该具有以下格式： void myCallback（const std:：string_view&tag）； 样例： channel.consume("myqueue").onSuccess( [](const std::string_view& tag) { std::cout << "Started consuming under tag "； std::cout << tag << std::endl; }); */ DeferredConsumer &consume(const std::string_view &queue,const std::string_view &tag,int flags,const Table &arguments);/** *确认接收到的消息 **消费者客户端对收到的消息进行确认应答**当在DeferredConsumer:：onReceived()方法中接收到消息时， 必须确认该消息， 以便RabbitMQ将其从队列中删除（除非使用noack选项消费）* *支持以下标志： * *-多条确认多条消息：之前传递的所有未确认消息也会得到确认 * *@param deliveryTag    消息的唯一delivery标签 *@param flags          可选标志 *@return bool */ bool ack(uint64_t deliveryTag, int flags=0);};class DeferredConsumer { /* 注册一个回调函数，该函数在消费者启动时被调用void onSuccess(const std::string &consumertag) */ DeferredConsumer &onSuccess(const ConsumeCallback& callback);/* 注册回调函数，用于接收到一个完整消息的时候被调用 void MessageCallback(const AMQP::Message &message,  uint64_t deliveryTag, bool redelivered) */ DeferredConsumer &onReceived(const MessageCallback& callback);/* Alias for onReceived() */ DeferredConsumer &onMessage(const MessageCallback& callback);/* 注册要在服务器取消消费者时调用的函数 void CancelCallback(const std::string &tag) */ DeferredConsumer &onCancelled(const CancelCallback& callback);};class Message : public Envelope{ const std::string &exchange();const std::string &routingkey();};class Envelope : public MetaData{ const char *body();  // 获取消息正文uint64_t bodySize(); // 获取消息正文大小};
}

2.ev

typedef struct ev_async 
{ EV_WATCHER (ev_async);EV_ATOMIC_T sent; /* private */ 
}ev_async; //break type 
enum 
{ EVBREAK_CANCEL = 0, /* undo unloop */ EVBREAK_ONE    = 1, /* unloop once */ EVBREAK_ALL    = 2  /* unloop all loops */ 
}; // 实例化并获取IO事件监控接口句柄
struct ev_loop *ev_default_loop (unsigned int flags EV_CPP (= 0));
# define EV_DEFAULT  ev_default_loop (0) // 开始运行IO事件监控, 这是一个阻塞接口
int  ev_run (struct ev_loop *loop);/* break out of the loop */
// 结束IO监控
// 如果在主线程进行ev_run(), 则可以直接调用,
// 如果在其他线程中进行ev_run(), 需要通过异步通知进行
void ev_break (struct ev_loop *loop, int32_t break_type) ;  void (*callback)(struct ev_loop *loop, ev_async *watcher, int32_t revents);// 初始化异步事件结构, 并设置回调函数
void ev_async_init(ev_async *w, callback cb);
// 启动事件监控循环中的异步任务处理
void ev_async_start(struct ev_loop *loop, ev_async *w); 
// 发送当前异步事件到异步线程中执行
void ev_async_send(struct ev_loop *loop, ev_async *w);

5.使用

1.publish.cc

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>int main()
{// 1.实例化底层网络通信框架的IO事件监控句柄auto *loop = EV_DEFAULT;// 2.实例化libEvHandler句柄 -> 将AMQP框架与事件监控关联起来AMQP::LibEvHandler handler(loop);// 3.实例化连接对象AMQP::Address address("amqp://root:SnowK8989@127.0.0.1:5672/");AMQP::TcpConnection connection(&handler, address);// 4.实例化信道对象AMQP::TcpChannel channel(&connection);// 5.声明交换机channel.declareExchange("test-exchange", AMQP::ExchangeType::direct).onError([](const char *message){ std::cout << "声明交换机失败: " << message << std::endl; }).onSuccess([](){ std::cout << "test-exchange 交换机创建成功" << std::endl; });// 6.声明队列channel.declareQueue("test-queue").onError([](const char *message){ std::cout << "声明队列失败: " << message << std::endl; }).onSuccess([](){ std::cout << "test-queue 队列创建成功" << std::endl; });// 7.针对交换机和队列进行绑定channel.bindQueue("test-exchange", "test-queue", "test-queue-key").onError([](const char *message){ std::cout << "test-exchange - test-queue 绑定失败: " \<< message << std::endl; }).onSuccess([](){ std::cout << "test-exchange - test-queue 绑定成功" << std::endl; });// 8.向交换机发布消息for (int i = 0; i < 5; ++i){std::string msg = "Hello SnowK-" + std::to_string(i);if(channel.publish("test-exchange", "test-queue-key", msg) == false){std::cout << "publish 失败" << std::endl;}}// 9.启动底层网络通信框架 -> 开启IOev_run(loop, 0);return 0;
}

2.consume.cc

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>void MessageCB(AMQP::TcpChannel* channel, const AMQP::Message& message, uint64_t deliveryTag, bool redelivered)
{std::string msg;msg.assign(message.body(), message.bodySize());// 不能这样使用, AMQP::Message后面没有存'\0'// std::cout << message << std::endl std::cout << msg << std::endl;channel->ack(deliveryTag);
}int main()
{// 1.实例化底层网络通信框架的IO事件监控句柄auto *loop = EV_DEFAULT;// 2.实例化libEvHandler句柄 -> 将AMQP框架与事件监控关联起来AMQP::LibEvHandler handler(loop);// 3.实例化连接对象AMQP::Address address("amqp://root:SnowK8989@127.0.0.1:5672/");AMQP::TcpConnection connection(&handler, address);// 4.实例化信道对象AMQP::TcpChannel channel(&connection);// 5.声明交换机channel.declareExchange("test-exchange", AMQP::ExchangeType::direct).onError([](const char *message){ std::cout << "声明交换机失败: " << message << std::endl; }).onSuccess([](){ std::cout << "test-exchange 交换机创建成功" << std::endl; });// 6.声明队列channel.declareQueue("test-queue").onError([](const char *message){ std::cout << "声明队列失败: " << message << std::endl; }).onSuccess([](){ std::cout << "test-queue 队列创建成功" << std::endl; });// 7.针对交换机和队列进行绑定channel.bindQueue("test-exchange", "test-queue", "test-queue-key").onError([](const char *message){ std::cout << "test-exchange - test-queue 绑定失败: " \<< message << std::endl; }).onSuccess([](){ std::cout << "test-exchange - test-queue 绑定成功"; });// 8.订阅消息对垒 -> 设置消息处理回调函数auto callback = std::bind(MessageCB, &channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);channel.consume("test-queue", "consume-tag").onReceived(callback).onError([](const char *message){ std::cout << "订阅 test-queue 队列消息失败: " << message << std::endl;exit(0); });// 9.启动底层网络通信框架 -> 开启IOev_run(loop, 0);return 0;
}

3.makefile

all: publish consume
publish: publish.ccg++ -o $@ $^ -lamqpcpp -lev -std=c++17
consume: consume.ccg++ -o $@ $^ -lamqpcpp -lev -std=c++17.PHONY:clean
clean:rm publish consume

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

相关文章

【云原生】容器方案 isula、containerd 基本功能测试

【云原生】容器方案 isula、containerd 基本功能测试

isula、containerd 基本功能测试 测试环境 树莓派 4BCPU: Cortex-A72 4CRAM: 8GSD卡: 128G C10 A1 U3OS: openEuler-22.03Docker: 20.10.17iSula: 2.0.7Containerd: 1.6.6 除 Docker 外没有集成网络组件&#xff0c;isula 与 containerd 均使用 host 网络测试。 测试情况 …
阅读更多...
Docker 教程三 （Ubuntu Docker安装）

Docker 教程三 （Ubuntu Docker安装）

Ubuntu Docker 安装 Docker Engine-Community 支持以下的 Ubuntu 版本&#xff1a; Xenial 16.04 (LTS)Bionic 18.04 (LTS)Cosmic 18.10Disco 19.04 其他更新的版本…… Docker Engine - Community 支持上 x86_64&#xff08;或 amd64&#xff09;armhf&#xff0c;arm64&am…
阅读更多...
部署 Docker harbor (httphttps)及使用

部署 Docker harbor (httphttps)及使用

部署 Docker harbor (http/https)及使用 官网下载docker harbor 和docker-compose 下载最新版本即可 https://github.com/goharbor/harbor https://github.com/docker/compose/releases 一.Docker harbor 配置http使用 1.解压harbor.tar # 解压至指定目录 [rootdocker ~]…
阅读更多...
无人机飞手执照培训，三类、四类傻傻分不清楚

无人机飞手执照培训，三类、四类傻傻分不清楚

无人机飞手执照培训中的三类和四类&#xff0c;主要依据无人机的空机重量进行区分&#xff0c;并对应不同的飞行权限和应用场景。以下是对这两类执照的详细解析&#xff1a; 一、无人机飞手执照的三类与四类定义 1. 三类执照&#xff1a; 定义&#xff1a;三类执照是指允许操…
阅读更多...
多代理强化学习综述：原理、算法与挑战

多代理强化学习综述：原理、算法与挑战

1. 引言 多代理强化学习&#xff08;Multi-Agent Reinforcement Learning, MARL&#xff09;是强化学习的一个重要分支&#xff0c;它将传统的单代理强化学习概念扩展到多代理环境中。在MARL中&#xff0c;多个代理通过与环境和其他代理的交互来学习最优策略&#xff0c;以在协…
阅读更多...
Go基础知识：切片

Go基础知识：切片

数组 Go 数组的大小是固定的&#xff0c;其长度是其类型的一部分&#xff08;[4]int并且[5]int是不同的、不兼容的类型&#xff09; var a [10]intb : [2]string{"Penn", "Teller"} b : [...]string{"Penn", "Teller"}package maini…
阅读更多...
无人机之交互系统篇

无人机之交互系统篇

一、系统构成 无人机交互系统通常由多个子系统组成&#xff0c;包括但不限于&#xff1a; 多模式人机交互装置&#xff1a;这是人机交互系统的基础层&#xff0c;通常包括计算机、局域网、传感器等设备&#xff0c;用于实现操作员与无人机之间的数据交互和指令传递。例如&…
阅读更多...
QD1-P8 HTML 格式化标签（font、pre、b、strong、i、u、del、s、sub、sup）

QD1-P8 HTML 格式化标签（font、pre、b、strong、i、u、del、s、sub、sup）

本节学习&#xff1a;HTML 格式化标签。 本节视频 www.bilibili.com/video/BV1n64y1U7oj?p8 ‍ 一、font 标签 用途&#xff1a;定义文本的字体大小、颜色和 face&#xff08;字体类型&#xff09;。 示例 <!DOCTYPE html> <html><head><meta cha…
阅读更多...
最新文章

Copyright @ 2022~2023