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