当前位置： 首页 > article >正文

AMQP-CPP二次封装

article 2024/11/14 4:08:11

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

文章目录

  • 前言
  • 一、封装思想
  • 二、类与接口介绍
    • 1.AMQP::Channel:信道类
    • 2.发布/订阅/确认应答
    • 3.消息类
    • 4.libev相关
  • 三.具体封装
    • 1.成员变量
    • 2.声明组件
    • 3.发布消息
    • 4.订阅队列消息
  • 四.总体代码

前言

AMQP-CPP他不提供底层网络通信的过程，也没有网络IO监控。他只是进行了内部的一个应用层协议的定义，以及根据 amqp 协议它内部的核心概念所支持的这些功能的一个支持。
所以它就需要我们自己用户自己来去来定义这块的一个网络底层的一个网络通信。
但是它将一些知名的网络通信框架集成到了它的这个呃它的 sdk 当中，也就是说如果我们需要使用某种某种知名的这种网络通信框架来进行一个底层网络通信，它也是支持的。人家已经把内部的这些最基础的一个适配功能是已经完成了
我们项目中使用的是libev这个网络库来进行rabbitmq客户端和rabbitmq服务器的一个通信。

一、封装思想

RabbitMQ客户端API接口的二次封装:为了更适合项目中的使用
封装思想:
1.将事件监控结构loop，通信连接connection，以及信道channel封装起来
2.提供创建交换机，队列并进行直接绑定的接口
3.提供向指定交换机发布消息的接口
4.提供订阅指定队列消息的接口

二、类与接口介绍

1.AMQP::Channel:信道类

这些都是基础的接口，都是channel提供的，其中创建一个channel需要传入一个连接。由于这些操作都是异步的，所以他的返回值就是类似于std::future这样的一个效果，不过我们这里是设置一个回调函数，针对他的onError和onSuccess设置一个回调函数，当响应返回时，根据响应的正确或失败执行不同的回到函数。

Channel(Connection *connection); 
Deferred &declareExchange()声明交换机
DeferredQueue &declareQueue()声明队列
Deferred &bindQueue();将交换机与队列进行关系绑定的功能

2.发布/订阅/确认应答

发布订阅，以及确认应答接口，也是channel提供的。

bool publish()发布消息
Deferred consumer &consume()订阅队列消息
bool ack()消费者客户端对收到的消息进行确认应答

3.消息类

我们传输的消息都是这个类定义的对象，需要注意的是这个获取消息正文返回值是一个const chat*,我们不能直接进行打印，因为消息中可能是二进制数据，数据中包含了\0，我们应该获取消息正文大小，在进行打印。

class Message:消息类
const char *body()获取消息正文
uint64 t bodySize()获取消息正文大小

4.libev相关

前面也介绍了AMQP-CPP没有实现底层网络通信，我们这里使用libev这个网络通信库来进行网络通信，以及网络IO的监控。

这里的ev_default_loop就是获取一个io事件监控对象，libev和libevent都是基于事件驱动的网络库，muduo就是借鉴了这俩个库的思想，这里的loop就相当于muduo中的eventloop。这里还为我们提供了一个宏，我们可以直接使用这个宏来进行loop的初始化。

Iibev使用到的结构体与接口
struct ev loop *ev_default_loop();实例化并获取I/O事件监控结构句柄
#define EV DEFAULT ev_default_loop (0);

这里的ev_run就是启动事件监控，相当于启动了epoll,这是一个阻塞的接口，它里面是一个死循环。因此我们需要单独用一个线程来执行这个ev_run。不要让他阻塞主线程。
还有一个接口时ev_break，就是退出事件监控。但是我们不能在主线程直接调用这个接口，我们通过异步通知给事件监控线程发送一个异步通知。

int ev_run (struct ev loop *loop);开始运行!/0事件监控，这是一个阻塞接口
void ev_break (struct ev loop *loop, int32 t break type); 结束//0监控
如果在当前线程进行ev run则可以直接调用，如果在其他线程中进行ev run
需要通过异步通知进行

这三个接口就是用于结束事件监控，发送异步通知的接口。
首先我们要定义一个ev_async这样的一个对象，并设置一个回调函数，这个回调函数中就会进行ev_break的调用。
我们需要在rabbitmq客户端析构时进行这样三个函数的调用，完成事件监控的一个释放。当时在项目中，我们的程序是不会释放的。

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);
发送当前异步事件到异步线程中执行

三.具体封装

1.成员变量

第一个变量ev_async就是用于发送异步任务，结束事件监控的一个对象。
第二个变量就是我们初始化的一个Loop事件监控对象
第三个对象就是我们将libev网络通信和AMQP-CPP进行一个绑定的，AMQP-CPP他不提供网络通信过程，但是他在内部适配了一些网络框架，这个对象相当于是吧libev和AMQP进行一个绑定。
第四个对象就是一个connection连接，
第五个对象channel，它是在连接上跟细化的一个概念，我们的所有操作都是channel提供的。
第五个参数就是一个线程，我们要在这个线程中开启事件监控。

struct ev_async _async_watcher;
struct ev_loop *_loop;
std::unique_ptr<AMQP::LibEvHandler> _handler;
std::unique_ptr<AMQP::TcpConnection> _connection;
std::unique_ptr<AMQP::TcpChannel> _channel;
std::thread _loop_thread;

2.声明组件

这个接口是我们给使用者调用的，通过这个接口就可以完成交换机，队列的创建，以及将交换机和队列绑定关系的创建。在我们项目中，只用到了一个直接交换的方式，所以我们这里的交换机类型默认是直接交换。

void declareComponents(const std::string &exchange,
   const std::string &queue,
   const std::string &routing_key = "routing_key",
   AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct) {
   _channel->declareExchange(exchange, echange_type)
       .onError([](const char *message) {
           LOG_ERROR("声明交换机失败：{}", message);
           exit(0);
       })
       .onSuccess([exchange](){
           LOG_ERROR("{} 交换机创建成功！", exchange);
       });
   _channel->declareQueue(queue)
       .onError([](const char *message) {
           LOG_ERROR("声明队列失败：{}", message);
           exit(0);
       })
       .onSuccess([queue](){
           LOG_ERROR("{} 队列创建成功！", queue);
       });
   //6. 针对交换机和队列进行绑定
   _channel->bindQueue(exchange, queue, routing_key)
       .onError([exchange, queue](const char *message) {
           LOG_ERROR("{} - {} 绑定失败：", exchange, queue);
           exit(0);
       })
       .onSuccess([exchange, queue, routing_key](){
           LOG_ERROR("{} - {} - {} 绑定成功！", exchange, queue, routing_key);
       });
}

3.发布消息

这是给发布客户端提供的一个接口，用于发布消息，用户只需要提供一个string,AMQP内部会生成一个Message对象。

bool publish(const std::string &exchange, 
     const std::string &msg, 
     const std::string &routing_key = "routing_key") {
     LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routing_key);
     bool ret = _channel->publish(exchange, routing_key, msg);
     if (ret == false) {
         LOG_ERROR("{} 发布消息失败：", exchange);
         return false;
     }
     return true;
 }

4.订阅队列消息

这是提供给订阅客户端使用的，需要指定要订阅的队列名，以及一个回调函数，这个回调函数就是在订阅的队列有消息时，就会进行消息的一个推送，当推送到订阅客户端后我们需要对消息进行一个处理。在channel.consume这个接口的返回值，是一个Defreed，它可以进行一个回调函数的设置，这里我们使用一个lambda,在这个lamdna中我们调用用户设置进来的回调函数，同时进行一个确认银达。

用户设置的这个回调函数签名如下，相当于用户只需要关系消息的起始地址，以及一个消息大小，然后用户更具消息大小获取指定地址的消息进行处理就行。

using MessageCallback = std::function<void(const char*, size_t)>;

void consume(const std::string &queue, const MessageCallback &cb) {
 LOG_DEBUG("开始订阅 {} 队列消息！", queue);
  _channel->consume(queue, "consume-tag")  //返回值 DeferredConsumer
  .onReceived([this, cb](const AMQP::Message &message, 
       uint64_t deliveryTag, 
       bool redelivered) {
       cb(message.body(), message.bodySize());
       _channel->ack(deliveryTag);
   })
   .onError([queue](const char *message){
       LOG_ERROR("订阅 {} 队列消息失败: {}", queue, message);
       exit(0);
   });
 }

四.总体代码

#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <string>
#include <thread>
#include <functional>
#include <memory>
#include "logger.hpp"

namespace bite_im{
class MQClient {
    public:
        using MessageCallback = std::function<void(const char*, size_t)>;
        using ptr = std::shared_ptr<MQClient>;
        MQClient(const std::string &user, 
            const std::string passwd,
            const std::string host) {
            _loop = EV_DEFAULT;
            _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
            //amqp://root:123456@127.0.0.1:5672/
            std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
            AMQP::Address address(url);
            _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
            _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());

            _loop_thread = std::thread([this]() {
                ev_run(_loop, 0);
            });
        }
        ~MQClient() {
             ev_async_init(&_async_watcher, watcher_callback);
             ev_async_start(_loop, &_async_watcher);
             ev_async_send(_loop, &_async_watcher);
             _loop_thread.join();
             _loop = nullptr;
        }
        void declareComponents(const std::string &exchange,
            const std::string &queue,
            const std::string &routing_key = "routing_key",
            AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct) {
            _channel->declareExchange(exchange, echange_type)
                .onError([](const char *message) {
                    LOG_ERROR("声明交换机失败：{}", message);
                    exit(0);
                })
                .onSuccess([exchange](){
                    LOG_ERROR("{} 交换机创建成功！", exchange);
                });
            _channel->declareQueue(queue)
                .onError([](const char *message) {
                    LOG_ERROR("声明队列失败：{}", message);
                    exit(0);
                })
                .onSuccess([queue](){
                    LOG_ERROR("{} 队列创建成功！", queue);
                });
            //6. 针对交换机和队列进行绑定
            _channel->bindQueue(exchange, queue, routing_key)
                .onError([exchange, queue](const char *message) {
                    LOG_ERROR("{} - {} 绑定失败：", exchange, queue);
                    exit(0);
                })
                .onSuccess([exchange, queue, routing_key](){
                    LOG_ERROR("{} - {} - {} 绑定成功！", exchange, queue, routing_key);
                });
        }
        bool publish(const std::string &exchange, 
            const std::string &msg, 
            const std::string &routing_key = "routing_key") {
            LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routing_key);
            bool ret = _channel->publish(exchange, routing_key, msg);
            if (ret == false) {
                LOG_ERROR("{} 发布消息失败：", exchange);
                return false;
            }
            return true;
        }
        void consume(const std::string &queue, const MessageCallback &cb) {
            LOG_DEBUG("开始订阅 {} 队列消息！", queue);
            _channel->consume(queue, "consume-tag")  //返回值 DeferredConsumer
                .onReceived([this, cb](const AMQP::Message &message, 
                    uint64_t deliveryTag, 
                    bool redelivered) {
                    cb(message.body(), message.bodySize());
                    _channel->ack(deliveryTag);
                })
                .onError([queue](const char *message){
                    LOG_ERROR("订阅 {} 队列消息失败: {}", queue, message);
                    exit(0);
                });
        }
    private:
        static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents) {
            ev_break(loop, EVBREAK_ALL);
        }
    private:
        struct ev_async _async_watcher;
        struct ev_loop *_loop;
        std::unique_ptr<AMQP::LibEvHandler> _handler;
        std::unique_ptr<AMQP::TcpConnection> _connection;
        std::unique_ptr<AMQP::TcpChannel> _channel;
        std::thread _loop_thread;
};
}

http://www.kler.cn/a/316316.html

相关文章：

  • ML 系列： 第 24 节 — 离散概率分布（泊松分布）
  • TensorRT基础知识
  • YUM 的使用
  • 【JavaEE进阶】导读
  • C# 集合与泛型
  • 使用ookii-dialogs-wpf在WPF选择文件夹时能输入路径
  • Llama 3.1 Omni：颠覆性的文本与语音双输出模型
  • Linux下文件下载中文乱码问题
  • C++单例模式代码实现与分析
  • Spring Boot实用小技巧5 - 第527篇
  • Leetcode面试经典150题-198.打家劫舍
  • 【Git使用】删除Github仓库中的指定文件/文件夹
  • Linux通过yum安装Docker
  • 5G 扬帆新质跃，技术蝶变开新篇-第七届“绽放杯”5G应用征集大赛 5G应用融合技术专题赛圆满收官
  • mysql性能优化-索引优化
  • 一天认识一个硬件之内存条
  • 1688国内店铺装修新版后台 放大效果代码生成1688店铺怎么装修1688平台
  • 通过解预测和机器学习促进蚁群优化
  • 用户态缓存：环形缓冲区（Ring Buffer）
  • Python 中的 Kombu 类库
  • 前端vue压缩静态图片，压缩gif动态图片
  • Anaconda配置pytorch的基本操作
  • Error when custom data is added to Azure OpenAI Service Deployment
  • Python办公自动化教程（001）：PDF内容提取
  • Junit与Spring Test简单使用
  • AI量化交易机器人开发