RabbitMq C++客户端的使用

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);
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.hpp

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

class MQClient
{
public:
    using ptr = std::shared_ptr<MQClient>;
    using MessageCallback = std::function<void(const char *, size_t)>;
    MQClient(const std::string &user, const std::string &password, const std::string &host)
    {
        // 1.实例化底层网络通信框架的IO事件监控句柄
        _loop = EV_DEFAULT;
        // 2.实例化LibEvHandler句柄，将AMQP框架与事件监控关联起来
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        // 3.实例化连接对象
        // amqp://root:2162627569@127.0.0.1:5672/
        std::string url = "amqp://" + user + ":" + password + "@" + host + "/";
        AMQP::Address address(url);
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
        // 4.实例化信道对象
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());
        // 5.启动底层网络通信框架，开启IO
        _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 exchange_type = AMQP::ExchangeType::direct)
    {
        // 声明交换机
        _channel->declareExchange(exchange, exchange_type)
            .onError([&exchange](const char *msg)
                     {
        LOG_ERROR("{}交换机创建失败：{}",exchange,msg);
        exit(1); })
            .onSuccess([&exchange]()
                       { LOG_INFO("{}交换机创建成功！", exchange); });
        // 声明队列
        _channel->declareQueue(queue)
            .onError([&queue](const char *msg)
                     {
        LOG_ERROR("{}队列创建失败：{}",queue,msg);
        exit(1); })
            .onSuccess([&queue]()
                       { LOG_INFO("{}队列创建成功！", queue); });
        // 6.绑定交换机和队列
        _channel->bindQueue(exchange, queue, routing_key)
            .onError([&exchange, &queue](const char *msg)
                     {
        LOG_ERROR("{} - {}绑定失败：{}",exchange,queue,msg);
        exit(1); })
            .onSuccess([&exchange, &queue]()
                       { LOG_INFO("{} - {}绑定成功！", exchange, queue); });
    }
    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-tags")
            .onReceived([this, &cb](const AMQP::Message &message, uint32_t deliveryTag, bool redelivered)
                        {
            cb(message.body(),message.bodySize());
            _channel->ack(deliveryTag); })
            .onError([&queue](const char *message)
                     {
                    LOG_ERROR("订阅 {} 队列消息失败: {}", queue, message);
                    exit(1); });
    }

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

consume.cc

#include "rabbitmq.hpp"
#include "logger.hpp"
#include <gflags/gflags.h>

DEFINE_string(user, "root", "rabbitmq访问用户名");
DEFINE_string(password, "2162627569", "rabbitmq访问密码");
DEFINE_string(host, "127.0.0.1:5672", "rabbitmq服务器地址信息 host:port");

DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");

void callback(const char *body, size_t sz)
{
    std::string msg;
    msg.assign(body, sz);
    LOG_DEBUG("{}", msg);
}
int main(int argc, char *argv[])
{
    google::ParseCommandLineFlags(&argc, &argv, true);
    init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);

    MQClient client(FLAGS_user, FLAGS_password, FLAGS_host);
    client.declareComponents("test-exchange", "test-queue", "test-queue-key");
    client.consume("test-queue", callback);
    std::this_thread::sleep_for(std::chrono::seconds(60));
    return 0;
}

publish.cc

#include "rabbitmq.hpp"
#include "logger.hpp"
#include <gflags/gflags.h>

DEFINE_string(user, "root", "rabbitmq访问用户名");
DEFINE_string(password, "2162627569", "rabbitmq访问密码");
DEFINE_string(host, "127.0.0.1:5672", "rabbitmq服务器地址信息 host:port");

DEFINE_bool(run_mode, false, "程序的运行模式，false-调试； true-发布；");
DEFINE_string(log_file, "", "发布模式下，用于指定日志的输出文件");
DEFINE_int32(log_level, 0, "发布模式下，用于指定日志输出等级");

void callback(const char *body, size_t sz)
{
    std::string msg;
    msg.assign(body, sz);
    LOG_DEBUG("{}", msg);
}
int main(int argc, char *argv[])
{
    google::ParseCommandLineFlags(&argc, &argv, true);
    init_logger(FLAGS_run_mode, FLAGS_log_file, FLAGS_log_level);

    MQClient client(FLAGS_user, FLAGS_password, FLAGS_host);
    client.declareComponents("test-exchange", "test-queue", "test-queue-key");
    for (int i = 0; i < 10; i++)
    {
        std::string msg = "hello world - " + std::to_string(i);
        bool ret = client.publish("test-exchange", msg, "test-queue-key");
        if (ret == false)
        {
            std::cout << "publish 失败！\n";
        }
    }
    std::this_thread::sleep_for(std::chrono::seconds(3));
    return 0;
}

makefile

all:publish consume
publish:publish.cc
	g++ -g -o $@ $^ -std=c++17 -lamqpcpp -lev -lspdlog -lfmt -lgflags
consume:consume.cc
	g++ -g -o $@ $^ -std=c++17 -lamqpcpp -lev -lspdlog -lfmt -lgflags

.PHONY:clean
clean:
	rm -f publish consume