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.ccg++ -g -o $@ $^ -std=c++17 -lamqpcpp -lev -lspdlog -lfmt -lgflags
consume:consume.ccg++ -g -o $@ $^ -std=c++17 -lamqpcpp -lev -lspdlog -lfmt -lgflags.PHONY:clean
clean:rm -f publish consume