如何使用AMQP-CPP向RabbitMQ正确发送自定义消息？

核心问题分析 #

你遇到的问题本质是AMQP-CPP的异步模型适配和资源复用问题：

• 重复创建连接/通道会极大降低效率，且不符合RabbitMQ的最佳实践
• 自定义消息发送失败通常是因为跨线程调用AMQP操作、通道未就绪就发送，或者没有在事件循环上下文内执行publish

解决方案：复用连接+线程安全的消息队列 #

下面给出一套可复用的实现方案，确保连接/通道只初始化一次，同时支持在任意线程发送自定义消息，且严格符合AMQP-CPP的异步线程模型。

1. 实现可复用的Handler类

该类继承AMQP::LibEvHandler，维护连接、通道，以及线程安全的待发送消息队列，通过ev_async唤醒事件循环处理消息发送：

#include <AMQP.h>
#include <ev.h>
#include <queue>
#include <mutex>
#include <string>
#include <iostream>

// 自定义消息结构体，存储发送所需的所有信息
struct RabbitMessage {
    std::string exchange;
    std::string routingKey;
    std::string body;
};

class RabbitHandler : public AMQP::LibEvHandler {
private:
    ev_loop* _loop;
    AMQP::TcpConnection* _conn = nullptr;
    AMQP::TcpChannel* _channel = nullptr;
    std::queue<RabbitMessage> _msgQueue;
    std::mutex _queueMutex;
    ev_async _asyncWatcher;

    // 事件循环回调：处理待发送消息
    static void asyncCallback(EV_P_ ev_async* w, int) {
        auto handler = static_cast<RabbitHandler*>(w->data);
        handler->_processMessages();
    }

    void _processMessages() {
        std::lock_guard<std::mutex> lock(_queueMutex);
        while (!_msgQueue.empty()) {
            auto msg = _msgQueue.front();
            _msgQueue.pop();
            
            // 确保通道处于可用状态（已完成队列/交换机声明）
            if (_channel && _channel->usable()) {
                _channel->publish(msg.exchange, msg.routingKey, msg.body);
            }
        }
    }

public:
    RabbitHandler(ev_loop* loop) : _loop(loop) {
        // 初始化异步唤醒器，用于跨线程触发事件循环处理消息
        ev_async_init(&_asyncWatcher, asyncCallback);
        _asyncWatcher.data = this;
        ev_async_start(_loop, &_asyncWatcher);
    }

    ~RabbitHandler() {
        ev_async_stop(_loop, &_asyncWatcher);
        delete _channel;
        delete _conn;
    }

    // 初始化RabbitMQ连接和通道（仅调用一次）
    void init(const std::string& amqpUri, const std::string& queueName) {
        AMQP::Address addr(amqpUri);
        _conn = new AMQP::TcpConnection(this, addr);
        _channel = new AMQP::TcpChannel(_conn);

        // 声明持久化队列（按需配置参数）
        _channel->declareQueue(queueName, AMQP::durable)
            .onSuccess([queueName]() {
                std::cout << "队列[" << queueName << "]声明完成，通道就绪" << std::endl;
            });

        // 监听连接断开事件
        _conn->onClosed([]() {
            std::cerr << "RabbitMQ连接已断开" << std::endl;
        });
    }

    // 线程安全的消息发送接口
    void send(const std::string& exchange, const std::string& routingKey, const std::string& body) {
        std::lock_guard<std::mutex> lock(_queueMutex);
        _msgQueue.push({exchange, routingKey, body});
        // 唤醒事件循环处理消息
        ev_async_send(_loop, &_asyncWatcher);
    }

    // 重写错误处理函数，便于排查问题
    void onError(AMQP::TcpConnection*, const char* msg) override {
        std::cerr << "连接错误: " << msg << std::endl;
    }

    void onError(AMQP::TcpChannel*, const char* msg) override {
        std::cerr << "通道错误: " << msg << std::endl;
    }
};

2. 主程序调用示例

启动事件循环，同时可以在任意线程调用send方法发送自定义消息：

#include <thread>
#include <chrono>

int main() {
    // 初始化libev事件循环
    ev_loop* loop = ev_default_loop(0);
    RabbitHandler handler(loop);

    // 初始化RabbitMQ连接（替换为你的实际URI和队列名）
    handler.init("amqp://guest:guest@localhost/", "my_custom_queue");

    // 模拟在子线程发送自定义消息
    std::thread sender([&handler]() {
        for (int i = 0; i < 10; ++i) {
            std::string msgBody = "自定义消息 " + std::to_string(i);
            std::cout << "准备发送: " << msgBody << std::endl;
            // 使用默认交换机，路由键为队列名
            handler.send("", "my_custom_queue", msgBody);
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    });

    // 启动事件循环（阻塞直到手动退出）
    ev_run(loop, 0);

    sender.join();
    return 0;
}

关键注意事项 #

1. 资源复用：连接和通道仅初始化一次，后续所有发送操作复用同一通道，避免重复创建的开销
2. 线程安全：AMQP-CPP的对象（连接、通道）不是线程安全的，必须通过ev_async确保publish操作在事件循环线程执行
3. 通道就绪检查：必须等待队列/交换机声明完成（onSuccess回调触发）后再发送消息，否则会导致消息丢失
4. 错误排查：重写onError方法可以快速定位连接或通道的问题，比如认证失败、队列不存在等

内容的提问来源于stack exchange，提问作者Deadpool