【RabbitMq C++】消息队列组件

1. RabbitMq介绍

RabbitMq - 消息队列组件:实现两个客户端主机之间消息传输的功能(发布&订阅)。

一端发布消息,一端订阅消息,消息就会被推送到订阅消息那一端然后进行处理。

RabbitMq遵守AMQP协议(标准的高级消息队列协议)

AMQP协议核心概念:交换机(交换机类型)、队列,绑定,消息。

两个客户端之间进行消息传输,一端产生消息另一端接收消息然后处理。按照以前的思想就是两个客户端直接进行网络通信socket,通过网络消息将一条消息发送给对方让对方进行处理,这是一种最基础数据传输过程。

但是这种消息传输是存在缺陷的!如果有一端连接断开了,那另一端消息到底还发不发,是等,还是将这条消息丢弃掉。如果一直等,新产生的消息又该怎么办,总不能一直存着。所以这种安全性是很低的。而且一对一这种客户端里面,通常数据的产生和数据的处理所消耗的时间是不成正比的。通常消息的处理消耗时间更多。

基于两端消息进行安全传输的需求,所以高级消息队列组件就产生了。两端不直接进行消息传输了。而是通过消息队列服务器来进行一个中间的数据转发功能。发布消息客户端将信息发布到服务器上,服务器在将这条消息推送给订阅消息队列客户端让它来进行处理。

但是针对一个高级消息队列设计的话,单纯一个只是做中间数据转发其实是不够的。我们希望它能在做中间数据转发更加灵活,在不同场景提供不同的功能。这个时候就有了AMQP的核心概念(交换机、队列、绑定、消息)。

消息队列服务器里面首先有一个交换机,它是用来处理数据转发逻辑功能模块。然后还有队列。订阅客户端连接服务器告诉服务器订阅那个队列。发布客户端进行消息发布并不是直接把消息发布到某个队列中,而是把信息发布到交换机,由交换机来决定把这条消息放到那个队列。决定了这条消息推送到那个订阅客户端哪里去进行处理。

交换机该把消息放到那一个队列中呢?这个时候就有了不同的交换机类型:

  1. 广播交换:当交换机收到消息,则将消息发布到所有绑定的队列中

交换机和队列都创建好了之后,会把交换机和队列进行关系绑定,也就是交换机和队列建立一个关联关系。而且会设置一个routing key(路由密钥:一定规则的字符串)用来标识这是一个放置什么类型消息的队列。

  1. 直接交换:根据消息中的binding_key与绑定的routing_key对比,一致则放到队列中
  1. 主题交换:使用binding_key与绑定的routing_key进行规则匹配,成功则放入队列

2. 安装RabbitMQ

sudo apt install rabbitmq-server
# 启动服务
sudo systemctl start rabbitmq-server.service
# 查看服务状态
sudo systemctl status rabbitmq-server.service

# 安装完成的时候默认有个用户 guest ,但是权限不够,要创建一个
# administrator 用户,才可以做为远程登录和发表订阅消息:

 #添加用户
sudo rabbitmqctl add_user root 123456
 #设置用户 tag
sudo rabbitmqctl set_user_tags root administrator
 #设置用户权限
sudo rabbitmqctl set_permissions -p / root "." "." ".*"

# RabbitMQ 自带了 web 管理界面,执行下面命令开启
sudo rabbitmq-plugins enable rabbitmq_management

访问 webUI 界面, 默认端口为 15672

至此RabbitMQ安装成功。

3. 安装 RabbitMQ 的 C++客户端库

我们这里使用 AMQP-CPP 库来编写客户端程序。

先安装libev网络通信库。在搭建RabbitMQ客户端的时候需要进行一个网络通信的事件监控。事件监控我们可以自己写poll,epoll但是太麻烦了。这里我们使用第三方网络通信框架。RabbitMQ对libevent、libev等等这些都支持。这里我们选择的是libvev。

sudo apt install libev-dev #libev 网络库组件
git clone https://gitee.***/iOceanPlus_Forked/AMQP-CPP.git
cd AMQP-CPP/
make 
make install

至此可以通过 AMQP-CPP 来操作 rabbitmq

4. AMQP-CPP 库的简单使用

AMQP-CPP 是用于与 RabbitMq 消息中间件通信的 c++库。它能解析从 RabbitMq
服务发送来的数据,也可以生成发向 RabbitMq 的数据包。AMQP-CPP 库不会向
RabbitMq 建立网络连接,所有的网络I/O由用户完成。

  • 当然,AMQP-CPP 提供了可选的网络层接口,它预定义了 TCP 模块,用户就不用自己实现网络IO,我们也可以选择 libevent、libev、libuv、asio 等异步通信组件,需要手动安装对应的组件。
  • AMQP-CPP 完全异步,没有阻塞式的系统调用,不使用线程就能够应用在高性能
    应用中。
  • 注意:它需要 c++17 的支持

4.1 使用

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

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

4.1.1 TCP 模式

  • 实现一个类继承自 AMQP::TcpHandler 类, 它负责网络层的 TCP 连接
  • 重写相关函数, 其中必须重写 monitor 函数
  • 在 monitor 函数中需要实现的是将 fd 放入 eventloop(select、epoll)中监控, 当 fd可写可读就绪之后, 调用 AMQP-CPP 的 connection->process(fd, flags)方法
#include <amqpcpp.h>
#include <amqpcpp/linux_tcp.h>
class MyTcpHandler : public AMQP::TcpHandler
{
   
   
    /**
     *AMQP 库在创建新连接时调用的方法
     *与处理程序相关联。这是对处理程序的第一次调用
     *@param connection 附加到处理程序的连接
     */
    virtual void onAttached(AMQP::TcpConnection *connection) override
    {
   
   
        //@todo
        // 添加您自己的实现,例如初始化事物
        // 以处理连接。
    }
    /**
     *当 TCP 连接时由 AMQP 库调用的方法
     *已经建立。调用此方法后,库
     *仍然需要设置可选的 TLS 层和
     *在 TCP 层的顶部建立 AMQP 连接。,这种方法
     *总是与稍后对 onLost()的调用配对。
     *@param connection 现在可以使用的连接
     */
    virtual void onConnected(AMQP::TcpConnection *connection) override
    {
   
   
        //@todo
        // 添加您自己的实现(可能不需要)
    }
    /**
    *在建立安全 TLS 连接时调用的方法。
    *这只对 amqps://连接调用。它允许您检查连接是否足够安全,以满足
   您的喜好
    *(例如,您可以检查服务器证书)。AMQP 协议仍然需要启动。
    *@param connection 已被保护的连接
    *@param ssl 来自 openssl 库的 ssl 结构
    *@return bool 如果可以使用连接,则为 True
    */
    virtual bool onSecured(AMQP::TcpConnection *connection, const SSL *ssl) override
    {
   
   
        //@todo
        // 添加您自己的实现,例如读取证书并检查它是否确实是您的
        return true;
    }
    /**
    *当登录尝试成功时由 AMQP 库调用的方法。在此之后,连接就可以使用
   了。
    *@param connection 现在可以使用的连接
    */
    virtual void onReady(AMQP::TcpConnection *connection) override
    {
   
   
        //@todo
        // 添加您自己的实现,例如通过创建一个通道实例,然后开始发布或使用
    }

    /**
    *该方法在服务器尝试协商检测信号间隔时调用,
    *并被覆盖以摆脱默认实现(否决建议的检测信号间隔),转而接受该间
   隔。
    *@param connection 发生错误的连接
    *@param interval 建议的间隔(秒)
    */
    virtual uint16_t onNegotiate(AMQP::TcpConnection *connection,uint16_t interval)
    {
   
   
        // 我们接受服务器的建议,但如果间隔小于一分钟,我们将使用一分钟的间隔
        if (interval < 60)
            interval = 60;
        //@todo
        // 在事件循环中设置一个计时器,
        // 如果在这段时间内没有发送其他指令,
        // 请确保每隔 interval 秒调用 connection->heartbeat()。
        // 返回我们要使用的间隔
        return interval;
    }
    /**
     * *发生致命错误时由 AMQP 库调用的方法
例如,因为无法识别从 RabbitMQ 接收的数据,或者基础连接丢失。
此调用之后通常会调用 onLost()(如果错误发生在 TCP 连接建立之
后)和 onDetached()。
*@param connection 发生错误的连接
*@param message 一条人类可读的错误消息
*/
    virtual void onError(AMQP::TcpConnection *connection, const char *message) override
    {
   
   
        //@todo
        // 添加您自己的实现,例如,通过向程序的用户报告错误并记录错误
    }
    /**
    *该方法在 AMQP 协议结束时调用的方法。这是调用 connection.close
   ()以正常关闭连接的计数器部分。请注意,TCP 连接此时仍处于活动状态,您还
   将收到对 onLost()和 onDetached()的调用
    @param connection AMQP 协议结束的连接
    */
    virtual void onClosed(AMQP::TcpConnection *connection) override
    {
   
   
        //@todo
        // 添加您自己的实现, 可能没有必要,
        // 但如果您想在 amqp 连接结束后立即执行某些操作,
        // 又不想等待 tcp 连接关闭,则这可能会很有用
    }
    /**
     *当 TCP 连接关闭或丢失时调用的方法。
     *如果同时调用了 onConnected(),则始终调用此方法
     *@param connection 已关闭但现在无法使用的连接
     */
    virtual void onLost(AMQP::TcpConnection *connection) override
    {
   
   
        //@todo
        // 添加您自己的实现(可能没有必要)
    }
    /**
    *调用的最终方法。这表示将不再对处理程序进行有关连接的进一步调
   用。
   *@param connection 可以被破坏的连接
*/
    virtual void onDetached(AMQP::TcpConnection *connection) override
    {
   
   
        
转载请说明出处内容投诉
CSS教程网 » 【RabbitMq C++】消息队列组件

发表评论

欢迎 访客 发表评论

一个令你着迷的主题!

查看演示 官网购买