一. RabbitMq的介绍

• RabbitMQ 是一个由 Erlang 语言编写的 开源消息中间件，实现了 AMQP（Advanced Message Queuing
  Protocol，高级消息队列协议）。

• 它的核心作用是：在分布式系统中传递消息，实现 解耦、异步、削峰填谷 等功能。

• RabbitMQ 既可以部署在本地，也可以运行在云端，支持多种编程语言客户端。

RabbitMQ 就像一个“快递中转站”，生产者把包裹交给它，它根据规则把包裹送到合适的收件人手里，保证安全、可靠、灵活。项目中用于消息队列的实现转发

二.AMQP-CPP 库的介绍

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

2.1AMQP-CPP的使用

AMQP-CPP 的使用有两种模式：

• 使用默认的TCP模块进行网络通信
• 使用拓展的libevent、libev、libuv、asio 异步通信组件进行通信

TCP模式

• 实现一个类继承自 AMQP::TcpHandler 类， 它负责网络层的 TCP 连接
• 重写相关函数， 其中必须重写 monitor 函数
• 在 monitor 函数中需要实现的是将 fd 放入 eventloop(select、 epoll)中监控， 当 fd 可写可读就绪之后， 调用 AMQP-CPP 的 connection->process(fd, flags)方法

拓展模式

以 libev 为例， 我们不必要自己实现 monitor 函数， 可以直接使用 AMQP::LibEvHandler

2.2常用类与接口介绍

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()： q
    };
    class Envelope : public MetaData
    {
        const char *body()
            uint64_t bodySize()
    };
}

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 */
};

struct ev_loop *ev_default_loop(unsigned int flags EV_CPP(= 0))
#define EV_DEFAULT ev_default_loop(0)

int ev_run(struct ev_loop *loop);
/* break out of the loop */
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);

三.二次封装

在项目中使用 rabbitmq 的时候，我们目前只需要交换机与队列的直接交换，实现一台主机将消息发布给另一台主机进行处理的功能，因此在这里可以对 mq 的操作进行简单的封装，使 mq 的操作在项目中更加简便：

封装一个 MQClient：

• 提供声明指定交换机与队列，并进行绑定的功能；
• 提供向指定交换机发布消息的功能
• 提供订阅指定队列消息，并设置回调函数进行消息消费处理的功能

#pragma once

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include "logger.hpp"

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);
            });
        // 针对交换机和队列进行绑定
        _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);
        });
    }

    // 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;
};