RabbitMq C++客户端的使用

namespace AMQP {
/**
* Generic callbacks that are used by many deferred objects
*/
using SuccessCallback = std::function<void()>;
using ErrorCallback = std::function<void(const char 
*message)>;
using FinalizeCallback = std::function<void()>;
/**
* Declaring and deleting a queue
*/
using QueueCallback = std::function<void(const std::string 
&name,uint32_t messagecount, uint32_t consumercount)>;
using DeleteCallback = std::function<void(uint32_t deletedmessages)>;using MessageCallback = std::function<void(const Message &message, uint64_t deliveryTag, bool redelivered)>;
//当使用发布者确认时，当服务器确认消息已被接收和处理时，将调用
AckCallback
using AckCallback = std::function<void(uint64_t deliveryTag, bool multiple)>;
//使用确认包裹通道时，当消息被 ack/nacked 时，会调用这些回调
using PublishAckCallback = std::function<void()>;
using PublishNackCallback = std::function<void()>;
using PublishLostCallback = std::function<void()>;
class Channel {Channel(Connection *connection);
bool connected()/***声明交换机*如果提供了一个空名称，则服务器将分配一个名称。*以下 flags 可用于交换机：* *-durable 持久化，重启后交换机依然有效*-autodelete 删除所有连接的队列后，自动删除交换*-passive 仅被动检查交换机是否存在*-internal 创建内部交换* *@param name 交换机的名称*@param-type 交换类型enum ExchangeType{ fanout, 广播交换，绑定的队列都能拿到消息direct, 直接交换，只将消息交给 routingkey 一致的队列topic, 主题交换，将消息交给符合 bindingkey 规则的队
列headers,consistent_hash,message_deduplication};*@param flags 交换机标志*@param arguments 其他参数* *此函数返回一个延迟处理程序。可以安装回调using onSuccess(), onError() and onFinalize() methods.*/Deferred &declareExchange(const std::string_view &name, ExchangeType type, int flags, const Table &arguments)/***声明队列*如果不提供名称，服务器将分配一个名称。*flags 可以是以下值的组合：* *-durable 持久队列在代理重新启动后仍然有效*-autodelete 当所有连接的使用者都离开时，自动删除队列*-passive 仅被动检查队列是否存在
*-exclusive 队列仅存在于此连接，并且在连接断开时自动删除* *@param name 队列的名称*@param flags 标志组合*@param arguments 可选参数* *此函数返回一个延迟处理程序。可以安装回调*使用 onSuccess（）、onError（）和 onFinalize（）方法。* Deferred &onError(const char *message)* *可以安装的 onSuccess（）回调应该具有以下签名：void myCallback(const std::string &name, uint32_t messageCount, uint32_t consumerCount);例如：channel.declareQueue("myqueue").onSuccess([](const std::string &name, uint32_t messageCount,uint32_t consumerCount) {std::cout << "Queue '" << name << "' ";std::cout << "has been declared with ";std::cout << messageCount;std::cout << " messages and ";std::cout << consumerCount;std::cout << " consumers" << std::endl;* });*/DeferredQueue &declareQueue(const std::string_view &name, int flags, const Table &arguments)/***将队列绑定到交换机* *@param exchange 源交换机*@param queue 目标队列*@param routingkey 路由密钥*@param arguments 其他绑定参数* *此函数返回一个延迟处理程序。可以安装回调*使用 onSuccess（）、onError（）和 onFinalize（）方法。*/Deferred &bindQueue(const std::string_view &exchange, const std::string_view &queue, const std::string_view &routingkey, const Table &arguments) /***将消息发布到 exchange*您必须提供交换机的名称和路由密钥。然后，RabbitMQ 将尝试将消息发送到一个或多个队列。使用可选的 flags 参数，可以指定如果消息无法路由到队列时应该发生
的情况。默认情况下，不可更改的消息将被静默地丢弃。* *如果设置了'mandatory'或'immediate'标志，则无法处理的消息将返回到应用程序。在开始发布之前，请确保您已经调用了 recall（）-方法，并设置了所有适当的处理程序来处理这些返回的消息。* *可以提供以下 flags：* *-mandatory 如果设置，服务器将返回未发送到队列的消息*-immediate 如果设置，服务器将返回无法立即转发给使用者的消息。*@param exchange 要发布到的交易所*@param routingkey 路由密钥*@param envelope 要发送的完整信封*@param message 要发送的消息*@param size 消息的大小*@param flags 可选标志*/bool publish(const std::string_view &exchange, const std::string_view &routingKey, const std::string &message, int flags = 0)/***告诉 RabbitMQ 服务器我们已准备好使用消息-也就是订阅队列消息* *调用此方法后，RabbitMQ 开始向客户端应用程序传递消息。consumer tag 是一个字符串标识符，如果您以后想通过 channel:：cancel（）调用停止它，
可以使用它来标识使用者。*如果您没有指定使用者 tag，服务器将为您分配一个。* *支持以下 flags：* *-nolocal 如果设置了，则不会同时消耗在此通道上发布的消息*-noack 如果设置了，则不必对已消费的消息进行确认*-exclusive 请求独占访问，只有此使用者可以访问队列* *@param queue 您要使用的队列*@param tag 将与此消费操作关联的消费者标记*@param flags 其他标记*@param arguments 其他参数* *此函数返回一个延迟处理程序。可以使用 onSuccess（）、onError（）和 onFinalize（）方法安装回
调。可以安装的 onSuccess（）回调应该具有以下格式：void myCallback（const std:：string_view&tag）；样例：channel.consume("myqueue").onSuccess([](const std::string_view& tag) {std::cout << "Started consuming under tag "；std::cout << tag << std::endl;});*/DeferredConsumer &consume(const std::string_view &queue, const std::string_view &tag, int flags, const Table &arguments)/***确认接收到的消息* *当在 DeferredConsumer:：onReceived（）方法中接收到消息时，必须确认该消息，以便 RabbitMQ 将其从队列中删除（除非使用 noack 选项消费）。* *支持以下标志：* *-多条确认多条消息：之前传递的所有未确认消息也会得到确认
**@param deliveryTag 消息的唯一 delivery 标签*@param flags 可选标志*@return bool*/bool ack(uint64_t deliveryTag, int flags=0)
} 
class DeferredConsumer {/*注册一个回调函数，该函数在消费者启动时被调用。void onSuccess(const std::string &consumertag)*/DeferredConsumer &onSuccess(const ConsumeCallback& callback)/*注册回调函数，用于接收到一个完整消息的时候被调用void MessageCallback(const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)*/DeferredConsumer &onReceived(const MessageCallback& callback)/* Alias for onReceived() */DeferredConsumer &onMessage(const MessageCallback& callback)/*注册要在服务器取消消费者时调用的函数void CancelCallback(const std::string &tag)*/DeferredConsumer &onCancelled(const CancelCallback& callback)
} 
class Message : public Envelope{const std::string &exchange()const std::string &routingkey()：q 
} 
class Envelope : public MetaData{const char *body() uint64_t bodySize()
} 
}

6.2ev

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);
/* break out of the 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) ;

7.简单使用

publish.cpp

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>int main()
{// 创建一个事件循环对象auto *loop = EV_DEFAULT;// 创建一个LibEvHandler对象AMQP::LibEvHandler handler(loop);// 设置RabbitMQ服务器的地址AMQP::Address address("amqp://root:123456@127.0.0.1:5672/");// 创建一个TcpConnection对象，连接到RabbitMQ服务器AMQP::TcpConnection connection(&handler, address);// 创建一个Channel对象，用于与RabbitMQ服务器进行通信AMQP::TcpChannel channel(&connection);// 声明一个交换机// 设置声明交换机失败时的回调函数channel.declareExchange("test_exchange", AMQP::direct)\.onError([](const std::string& message){std::cout << "Error1: " << message << std::endl;return -1;})\.onSuccess([](){std::cout << "Exchange declared" << std::endl;});channel.declareQueue("test_queue")\.onError([](const std::string& message){std::cout << "Error2: " << message << std::endl;return -1;})\.onSuccess([](){std::cout << "Queue declared" << std::endl;});// 绑定队列到交换机channel.bindQueue("test_exchange", "test_queue", "test_queue")\.onError([](const std::string& message){std::cout << "Error2: " << message << std::endl;return -1;})\.onSuccess([](){std::cout << "Queue bind" << std::endl;});// 发送10条消息for(int i = 1; i <= 10; i++){std::string message = "Hello RabbitMQ " + std::to_string(i);// 发布消息到RabbitMQ服务器bool ret = channel.publish("test_exchange", "test_queue", message);if (!ret){std::cout << "Publish failed" << std::endl;return -1;}}// 运行事件循环ev_run(loop, 0);return 0;
}

consume.cpp

#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>void MessageCb(AMQP::TcpChannel *channel, const AMQP::Message &message,uint64_t deliveryTag, bool redelivered)
{std::string body = message.body();std::cout << "Received a message with content: " << body << std::endl;channel->ack(deliveryTag);
}int main()
{// 创建一个事件循环对象auto *loop = EV_DEFAULT;// 创建一个LibEvHandler对象AMQP::LibEvHandler handler(loop);// 设置RabbitMQ服务器的地址AMQP::Address address("amqp://root:123456@127.0.0.1:5672/");// 创建一个TcpConnection对象，连接到RabbitMQ服务器AMQP::TcpConnection connection(&handler, address);// 创建一个Channel对象，用于与RabbitMQ服务器进行通信AMQP::TcpChannel channel(&connection);// 声明一个交换机// 设置声明交换机失败时的回调函数channel.declareExchange("test_exchange", AMQP::direct)\.onError([](const std::string& message){std::cout << "Error: " << message << std::endl;return -1;})\.onSuccess([](){std::cout << "Exchange declared" << std::endl;});// 绑定队列到交换机channel.bindQueue("test_exchange", "test_queue", "test_queue")\.onError([](const std::string& message){std::cout << "Error: " << message << std::endl;return -1;})\.onSuccess([](){std::cout << "Queue bound" << std::endl;});auto call_back=std::bind(MessageCb,&channel,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);channel.consume("test_queue")\.onReceived(call_back)\.onError([](const std::string& message){std::cout << "Error: " << message << std::endl;return -1;});// 运行事件循环ev_run(loop, 0);return 0;
}

makefile

all : publish consume
publish:publish.cppg++ publish.cpp -o publish -std=c++17 -lamqpcpp -lev -lpthread -ldl -lssl -lcrypto
consume:consume.cppg++ consume.cpp -o consume -std=c++17 -lamqpcpp -lev -lpthread -ldl -lssl -lcrypto@PHONY:clean
clean:rm -f publish consume

验证截图

至此大家就可以简单安全和使用RabbitMq!