C++ RabbitMQ封装

左手的ㄟ右手 2022-02-26 02:28 715阅读 0赞

### C++ RabbitMQ封装 ###

RabbitMq的源码可以在官网下载https://www.rabbitmq.com/，需要使用cmake转为vs工程。由于源码不大，此处选择编译librabbitmq，生成librabbitmq.4.lib静态库，直接链接到自己的程序中。  
附上vs版的源码 RabbitMq VS2017工程，我使用的是vs2017，低版本的可以自行百度修改vs版本。

首先封装CQueue类，包含队列的持久化，自动创建，自动删除，队列名等属性

/**
    *   @brief 消息队列
    **/
    class CQueue
    {
    public:
    	CQueue(std::string strName, bool nDurable = true, bool nPassive = false,
    		bool bAutoDelete = false, bool bExclusive = false)
    		:m_name(strName),
    		 m_durable(nDurable),
    		 m_bPassive(nPassive),
    		 m_bAutoDelete(bAutoDelete),
    		 m_bExclusive(bExclusive)
    	{
    		
    	}
    	
    	CQueue(const CQueue &other)
    	{
    		this->m_name = other.m_name;
    		this->m_durable = other.m_durable;
    		this->m_bAutoDelete = other.m_bAutoDelete;
    		this->m_bExclusive = other.m_bExclusive;
    		this->m_bPassive = other.m_bPassive;
    	}
    	
    	CQueue operator=(const CQueue &other)
    	{
    		if (this == &other) 
    			return *this;
    		this->m_name = other.m_name;
    		this->m_durable = other.m_durable;
    		this->m_bAutoDelete = other.m_bAutoDelete;
    		this->m_bExclusive = other.m_bExclusive;
    		this->m_bPassive = other.m_bPassive;
    		return *this;
    	}
    
    	/**
    	* m_name 消息队列名称
    	*
    	**/
    	std::string   m_name;
    
    	/**
    	* m_durable 队列是否持久化(当mq服务端断开重启后，队列是否还存在)
    	*
    	**/
    	bool		  m_durable;
    
    	/**
    	* m_exclusive 是否是专用队列（当前连接不在时，队列是否删除）
    	*		如果m_exclusive = 1，那么当当前连接断开时，队列也被删除
    	*		否则，当当前连接断开时，队列依旧存在
    	**/
    	bool		  m_bExclusive;
    
    
    	/**
    	* m_auto_delete 没有consumer时，队列是否自动删除
    	*
    	**/
    	bool		  m_bAutoDelete;
    
    	/**
    	* passive 检测queue是否存在
    	*	设为true，
    	*		若queue存在则创建命令成功返回（调用其他参数不会影响queue属性），
    	*		若不存在不会创建queue，返回错误。
    	*	设为false，
    	*		如果queue不存在则创建，调用成功返回。
    	*		如果queue已经存在，并且匹配现在queue的话则成功返回，如果不匹配则queue声明失败。
    	**/
    	bool		  m_bPassive;
    };

其次封装CExchange类，包含交换机的持久化，自动创建，自动删除，交换机名等属性

/**
    *   @brief 交换机
    **/
    class CExchange
    {
    public:
    	CExchange(std::string name, std::string type = "direct",
    		bool durable = true, bool passive = false, bool auto_delete = false, bool internal = false)
    		:m_name(name), 
    		 m_type(type),
    		 m_bDurable(durable), 
    		 m_bPassive(passive),
    		 m_bAutoDelete(auto_delete),
    		 m_bInternal(internal)
    	{
    		
    	}
    	
    	CExchange(const CExchange &other)
    	{
    		this->m_name = other.m_name;
    		this->m_bDurable = other.m_bDurable;
    		this->m_type = other.m_type;
    		this->m_bAutoDelete = other.m_bAutoDelete;
    		this->m_bInternal = other.m_bInternal;
    		this->m_bPassive = other.m_bPassive;
    	}
    	
    	CExchange operator=(const CExchange &other)
    	{
    		if (this == &other) 
    			return *this;
    		this->m_name = other.m_name;
    		this->m_bDurable = other.m_bDurable;
    		this->m_type = other.m_type;
    		this->m_bAutoDelete = other.m_bAutoDelete;
    		this->m_bInternal = other.m_bInternal;
    		this->m_bPassive = other.m_bPassive;
    		return *this;
    	}
    
    	/**
    	* m_name 交换机名称
    	*
    	**/
    	std::string   m_name;
    
    	/**
    	* m_type 指定exchange类型，"fanout"  "direct" "topic"三选一
    	*	"fanout" 广播的方式，发送到该exchange的所有队列上(不需要进行bind操作)
    	*	"direct" 通过路由键发送到指定的队列上(把消息发送到匹配routing key的队列中。)
    	*	"topic" 通过匹配路由键的方式获取，使用通配符*，#
    	**/
    	std::string   m_type;
    
    	/**
    	* m_durable 交换机是否持久化(当mq服务端断开重启后，交换机是否还存在)
    	**/
    	bool		  m_bDurable;  
    
    	/**
    	* m_auto_delete 连接断开时，交换机是否自动删除
    	*
    	**/
    	bool		  m_bAutoDelete;
    
    	/**
    	* m_internal 默认为0，没有使用到
    	*
    	**/
    	bool		  m_bInternal;
    
    	/**
    	* passive 检测exchange是否存在
    	*	设为true，
    	*		若exchange存在则命令成功返回（调用其他参数不会影响exchange属性），
    	*		若不存在不会创建exchange，返回错误。
    	*	设为false，
    	*		如果exchange不存在则创建exchange，调用成功返回。
    	*		如果exchange已经存在，并且匹配现在exchange的话则成功返回，如果不匹配则exchange声明失败。
    	**/
    	bool		  m_bPassive;
    };

再封装一个消息类，包含消息的具体内容，消息属性，消息路由名等属性：

/**
    *   @brief 消息结构
    **/
    class CMessage
    {
    public:
    	CMessage(std::string data, amqp_basic_properties_t properties, std::string  routekey, 
    		bool bMandatory = true, bool bImmediate = false)
    		:m_data(data), 
    		 m_properties(properties),
    		 m_routekey(routekey),
    		 m_bMandatory(bMandatory),
    		 m_bImmediate(bImmediate)
    	{
    		
    	}
    	
    	CMessage(const CMessage &other)
    	{
    		this->m_data = other.m_data;
    		this->m_properties = other.m_properties;
    		this->m_routekey = other.m_routekey;
    		this->m_bMandatory = other.m_bMandatory;
    		this->m_bImmediate = other.m_bImmediate;
    	}
    	
    	CMessage operator=(const CMessage &other)
    	{
    		if (this == &other) 
    			return *this;
    		this->m_data = other.m_data;
    		this->m_properties = other.m_properties;
    		this->m_routekey = other.m_routekey;
    		this->m_bMandatory = other.m_bMandatory;
    		this->m_bImmediate = other.m_bImmediate;
    		return *this;
    	}
    
    	/**
    	* m_bMandatory 消息必须路由到队列，由消费者来取
    	*
    	**/
    	bool m_bMandatory;
    
    	/**
    	* m_bImmediate 消息是否要立即发送到消费者
    	*
    	**/
    	bool m_bImmediate;
    
    	/**
    	* m_routekey 消息路由 默认所有消息的路由都以 “msg.”开头
    	*
    	**/
    	std::string  m_routekey;
    
    	/**
    	* m_data 消息内容（目前RabbitMq只支持字符串，如要传json，只需写json格式即可，{\"test\":\"hello\"}）
    	*
    	**/
    	std::string  m_data;
    
    	/**
    	* m_properties 消息的属性(包括消息头)
    	*
    	**/
    	amqp_basic_properties_t m_properties;
    };

最后就是RabbitMQ操作封装，包含mq的连接与断开，交换机和队列的声明，交换机和队列的绑定，发布消息，消费消息等操作：

/**
    *   @brief RabbitMq封装类
    **/
    class CRabbitMqClient
    {
    public:
    	CRabbitMqClient(
    		std::string HostName = connect_host,
    		uint32_t port = connect_port,
    		std::string usr = connect_user,
    		std::string psw = connect_pwd
    	);
    	~CRabbitMqClient();
    
    public:
    	/**
    	* 连接RabbitMq服务器
    	*
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t Connect(std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 断开与RabbitMq服务器的连接
    	*
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t DisConnect(std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 交换机初始化
    	*
    	* @param[in]    exchange		交换机实例
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t DeclareExchange(CExchange &exchange, std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 队列初始化
    	*
    	* @param[in]    queue			消息队列实例
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t DeclareQueue(CQueue &queue, std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 将指定队列绑定到指定交换机上
    	*
    	* @param[in]    queue			消息队列实例
    	* @param[in]    exchange		交换机实例
    	* @param[in]    bind_key		路由名
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t BindQueueToExchange(CQueue &queue, CExchange &exchange, const std::string bind_key, std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 解绑指定队列和指定交换机
    	*
    	* @param[in]    queue			消息队列实例
    	* @param[in]    exchange		交换机实例
    	* @param[in]    bind_key		路由名
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t UnBindQueueToExchange(CQueue &queue, CExchange &exchange, const std::string bind_key, std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 发布单个消息
    	*
    	* @param[in]    message			消息实例
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t PublishMessage(const CMessage &message, std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 发布批量消息
    	*
    	* @param[in]    messageVec		消息实例数组
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t PublishMessage(std::vector<CMessage> &messageVec, std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 从指定队列中取单个消息（同步操作）
    	*
    	* @param[in]    queue_name_		队列名称
    	* @param[in]    timeout			超时时间，为NULL有可能导致阻塞
    	* @param[out]   message			消息内容
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t ConsumerMessage(std::string &message, const std::string &queue_name_, timeval timeout = { 5, 0 }, std::string &ErrorReturn = std::string(""));
    
    	/**
    	* 从指定队列中取指定数目消息（同步操作）
    	*
    	* @param[in]    queue_name_		队列名称
    	* @param[in]    GetNum			消息树木
    	* @param[in]    timeout			超时时间，为NULL有可能导致阻塞
    	* @param[out]   messageVec		消息内容数组（大小不一定等于GetNum）
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t ConsumerMessage(std::vector<std::string> &messageVec, const std::string &queue_name_, uint32_t GetNum = 1000, timeval timeout = { 5, 0 }, std::string &ErrorReturn = std::string(""));
    
    	//还有一个amqp_basic_get函数，没有用到，暂时不实现（区别于amqp_basic_consume，get是检索，consume是消费）
    private:
    	/**
    	* 获取错误信息文本
    	*
    	* @param[in]    reply			amqp_rpc_reply_t对象
    	* @param[in]    strModule		模块信息（一般传当前调用的RabbitMQ-API函数名）
    	* @param[out]   ErrorReturn		错误信息
    	* @return		int32_t			0表示成功，其他失败
    	*/
    	int32_t GetErrorText(const amqp_rpc_reply_t &reply, const std::string &strModule, std::string &ErrorReturn);
    
    private:
    	std::string m_strHost;
    	std::string m_strUser;
    	std::string m_strPwd;
    	int32_t m_nPort;
    
    	amqp_connection_state_t m_connectState;
    };

实现各个接口的cpp文件如下：

#include "RabbitMqClient.h"
    
    CRabbitMqClient::CRabbitMqClient(std::string HostName /*= CONNECT_HOST*/, uint32_t port /*= CONNECT_PORT*/,
    	std::string usr /*= CONNECT_USER*/, std::string psw /*= CONNECT_PWD */)
    	:m_strHost(HostName),
    	m_nPort(port),
    	m_strUser(usr),
    	m_strPwd(psw),
    	m_connectState(NULL)
    {
    
    }
    
    CRabbitMqClient::~CRabbitMqClient()
    {
    	DisConnect();
    }
    
    int32_t CRabbitMqClient::Connect(std::string &ErrorReturn /*= ""*/)
    {
    	m_connectState = amqp_new_connection();
    	if (!m_connectState)
    	{
    		ErrorReturn = "amqp_new_connection failed";
    		return -1;
    	}
    
    	amqp_socket_t *socket = amqp_tcp_socket_new(m_connectState);
    	if (!socket)
    	{
    		ErrorReturn = "amqp_tcp_socket_new failed";
    		return -1;
    	}
    
    	int rc = amqp_socket_open(socket, m_strHost.c_str(), AMQP_PROTOCOL_PORT);
    	if (rc != AMQP_STATUS_OK)
    	{
    		ErrorReturn = "amqp_socket_open failed";
    		return -1;
    	}
    
    	amqp_rpc_reply_t rpc_reply = amqp_login(
    		m_connectState, 
    		"/", 
    		1, 
    		AMQP_DEFAULT_FRAME_SIZE,
    		AMQP_DEFAULT_HEARTBEAT, 
    		AMQP_SASL_METHOD_PLAIN, 
    		m_strUser.c_str(), 
    		m_strPwd.c_str()
    	);
    	if (rpc_reply.reply_type != AMQP_RESPONSE_NORMAL)
    	{
    		ErrorReturn = "amqp_login failed";
    		return -1;
    	}
    
    	amqp_channel_open_ok_t *res =
    		amqp_channel_open(m_connectState, channel_id);
    	if (!res)
    	{
    		ErrorReturn = "amqp_channel_open failed";
    		return -1;
    	}
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::DisConnect(std::string &ErrorReturn /*= std::string("")*/)
    {
    	amqp_rpc_reply_t rpc_reply = amqp_channel_close(m_connectState, channel_id, AMQP_REPLY_SUCCESS);
    	if (rpc_reply.reply_type != AMQP_RESPONSE_NORMAL)
    	{
    		ErrorReturn = "amqp_channel_close failed";
    		return -1;
    	}
    
    	rpc_reply = amqp_connection_close(m_connectState, AMQP_REPLY_SUCCESS);
    	if (rpc_reply.reply_type != AMQP_RESPONSE_NORMAL)
    	{
    		ErrorReturn = "amqp_connection_close failed";
    		return -1;
    	}
    
    	int rc = amqp_destroy_connection(m_connectState);
    	if (rc != AMQP_STATUS_OK)
    	{
    		ErrorReturn = "amqp_destroy_connection failed";
    		return -1;
    	}
    	m_connectState = NULL;
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::DeclareExchange(CExchange &exchange, std::string &ErrorReturn /*= std::string("")*/)
    {
    	amqp_exchange_declare_ok_t *res = amqp_exchange_declare(
    		m_connectState,
    		channel_id,
    		amqp_cstring_bytes(exchange.m_name.c_str()),
    		amqp_cstring_bytes(exchange.m_type.c_str()),
    		exchange.m_bPassive,
    		exchange.m_bDurable,
    		exchange.m_bAutoDelete,
    		exchange.m_bInternal,
    		amqp_empty_table);
    	amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    	if (0 != GetErrorText(rpc_reply, "amqp_exchange_declare", ErrorReturn))
    	{
    		return -1;
    	}
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::DeclareQueue(CQueue &queue, std::string &ErrorReturn /*= std::string("")*/)
    {
    	amqp_queue_declare_ok_t *res = amqp_queue_declare(
    		m_connectState, 
    		channel_id, 
    		amqp_cstring_bytes(queue.m_name.c_str()),
    		queue.m_bPassive,	/*passive*/ 	//检验队列是否存在（同exchange中的passive属性）
    		queue.m_durable,	/*durable*/ 	//队列是否持久化(即使mq服务重启也会存在)
    		queue.m_bExclusive,  /*exclusive*/ 	//是否是专用队列（当前连接不在时，队列是否删除）
    		queue.m_bAutoDelete,/*auto_delete*/  //是否自动删除（什么时候删除？。。。）
    		amqp_empty_table);
    	amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    	if (0 != GetErrorText(rpc_reply, "amqp_queue_declare", ErrorReturn))
    	{
    		return -1;
    	}
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::BindQueueToExchange(CQueue &queue, CExchange &exchange, const std::string bind_key, std::string &ErrorReturn /*= std::string("")*/)
    {
    	//队列与交换机绑定
    	amqp_queue_bind_ok_t *res = amqp_queue_bind(
    		m_connectState,		//连接
    		channel_id,		//频道id
    		amqp_cstring_bytes(queue.m_name.c_str()),	//队列
    		amqp_cstring_bytes(exchange.m_name.c_str()),	//交换机
    		amqp_cstring_bytes(bind_key.c_str()), //路由
    		amqp_empty_table);
    	amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    	if (0 != GetErrorText(rpc_reply, "amqp_queue_bind", ErrorReturn))
    	{
    		return -1;
    	}
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::UnBindQueueToExchange(CQueue &queue, CExchange &exchange, const std::string bind_key, std::string &ErrorReturn /*= std::string("")*/)
    {
    	//队列与交换机绑定
    	amqp_queue_unbind_ok_t *res = amqp_queue_unbind(
    		m_connectState,		//连接
    		channel_id,		//频道id
    		amqp_cstring_bytes(queue.m_name.c_str()),	//队列
    		amqp_cstring_bytes(exchange.m_name.c_str()),	//交换机
    		amqp_cstring_bytes(bind_key.c_str()), //路由
    		amqp_empty_table);
    	amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    	if (0 != GetErrorText(rpc_reply, "amqp_queue_unbind", ErrorReturn))
    	{
    		return -1;
    	}
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::PublishMessage(const CMessage &message, std::string &ErrorReturn /*= std::string("")*/)
    {
    	int retval = amqp_basic_publish(
    		m_connectState,
    		channel_id,		//频道id，没有要求，用同一个频道是ok的
    		amqp_cstring_bytes(exchange_name),	//交换机
    		amqp_cstring_bytes(routing_keys_name),	//路由
    		message.m_bMandatory,	//消息必须路由到队列，由消费者来取
    		message.m_bImmediate,	//消息是否要立即发送到消费者
    		&message.m_properties,	//消息的属性，支持key-value模式
    		amqp_cstring_bytes(message.m_data.c_str()) //消息内容
    	);
    	if (retval != AMQP_STATUS_OK)
    	{
    		amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    		if (0 != GetErrorText(rpc_reply, "amqp_basic_publish", ErrorReturn))
    		{
    			return -1;
    		}
    	}
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::PublishMessage(std::vector<CMessage> &messageVec, std::string &ErrorReturn /*= std::string("")*/)
    {
    	int ret = 0;
    
    	for (std::vector<CMessage>::iterator it = messageVec.begin(); it != messageVec.end(); ++it)
    	{
    		//发送多个消息时，暂时简化为只看发送最后一条消息的结果
    		ret = PublishMessage(*it, ErrorReturn);
    	}
    
    	return ret;
    }
    
    int32_t CRabbitMqClient::ConsumerMessage(std::string &message, const std::string &queue_name_, timeval timeout/* = { 5, 0 }*/, std::string &ErrorReturn/* = std::string("")*/)
    {
    	//step 1
    	amqp_basic_qos_ok_t *res = amqp_basic_qos(
    		m_connectState,	//MQ连接
    		channel_id,	//频道号
    		0,		//这个参数从rabbitmq中没用到，默认为0
    		1,		//设置为rabbitmq流量控制数量，取消息数目
    		0		//glotal=true时表示在当前channel上所有的consumer都生效，否则只对设置了之后新建的consumer生效
    	);
    	if (!res)
    	{
    		amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    		if (0 != GetErrorText(rpc_reply, "amqp_basic_qos", ErrorReturn))
    		{
    			return -1;
    		}
    	}
    
    	//step2
    	amqp_basic_consume_ok_t *res2 =amqp_basic_consume(
    			m_connectState, 
    			channel_id,
    			amqp_cstring_bytes(queue_name_.c_str()), 
    			amqp_empty_bytes,
    			/*no_local*/ 0,	//true if the server should not deliver to this consumer messages published on this channel’s connection
    							//true：mq服务器不应将当前频道上的消息发送到此consumer
    			/*no_ack*/ 1,	//是否需要确认消息后再从队列中删除消息
    			/*exclusive*/ 0, 
    			amqp_empty_table);
    	if (!res)
    	{
    		amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    		if (0 != GetErrorText(rpc_reply, "amqp_basic_consume", ErrorReturn))
    		{
    			return -1;
    		}
    	}
    
    	//step3
    	amqp_envelope_t envelope;
    	amqp_rpc_reply_t rpc_reply =
    		amqp_consume_message(m_connectState, &envelope, &timeout, 0);
    	if (0 != GetErrorText(rpc_reply, "amqp_consume_message", ErrorReturn))
    	{
    		return -1;
    	}
    
    	message = std::string((char *)envelope.message.body.bytes, (char *)envelope.message.body.bytes + envelope.message.body.len);
    	amqp_destroy_envelope(&envelope);
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::ConsumerMessage(std::vector<std::string> &messageVec, const std::string &queue_name_,
    	uint32_t GetNum/* = 1000*/, timeval timeout/* = { 5, 0 }*/, std::string &ErrorReturn/* = std::string("")*/)
    {
    	//step 1
    	amqp_basic_qos_ok_t *res = amqp_basic_qos(
    		m_connectState,
    		channel_id,
    		0,
    		GetNum,
    		0);
    	if (!res)
    	{
    		amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    		if (0 != GetErrorText(rpc_reply, "amqp_basic_qos", ErrorReturn))
    		{
    			return -1;
    		}
    	}
    
    	//step2
    	amqp_basic_consume_ok_t *res2 = amqp_basic_consume(
    		m_connectState,
    		channel_id,
    		amqp_cstring_bytes(queue_name_.c_str()),
    		amqp_empty_bytes,
    		/*no_local*/ 0,
    		/*no_ack*/ 1,
    		/*exclusive*/ 0,
    		amqp_empty_table);
    	if (!res)
    	{
    		amqp_rpc_reply_t rpc_reply = amqp_get_rpc_reply(m_connectState);
    		if (0 != GetErrorText(rpc_reply, "amqp_basic_consume", ErrorReturn))
    		{
    			return -1;
    		}
    	}
    
    	while (GetNum--)
    	{
    		amqp_envelope_t envelope;
    		amqp_rpc_reply_t rpc_reply =
    			amqp_consume_message(m_connectState, &envelope, &timeout, 0);
    		if (0 != GetErrorText(rpc_reply, "amqp_consume_message", ErrorReturn))
    		{
    			return -1;
    		}
    
    		size_t body_size = envelope.message.body.len;
    		char *body = (char*)malloc(body_size + 1);
    		memset(body, 0, body_size + 1);
    		if (body)
    		{
    			memcpy(body, envelope.message.body.bytes, body_size);
    		}
    		std::string message = body;
    		messageVec.push_back(message);
    		free(body);
    		amqp_destroy_envelope(&envelope);
    	}
    
    	return 0;
    }
    
    int32_t CRabbitMqClient::GetErrorText(const amqp_rpc_reply_t &reply, const std::string &strModule, std::string &ErrorReturn)
    {
    	char rtnmsg[1024] = { 0 };
    	switch (reply.reply_type)
    	{
    	case AMQP_RESPONSE_NORMAL:
    		return 0;
    	case AMQP_RESPONSE_NONE:
    		sprintf(rtnmsg, "%s: missing RPC reply type!\n", strModule.c_str());
    		break;
    	case AMQP_RESPONSE_LIBRARY_EXCEPTION:
    		sprintf(rtnmsg, "%s: %s\n", strModule.c_str(), amqp_error_string2(reply.library_error));
    		break;
    	case AMQP_RESPONSE_SERVER_EXCEPTION:
    		switch (reply.reply.id)
    		{
    			case AMQP_CONNECTION_CLOSE_METHOD: 
    				{
    					amqp_connection_close_t *m = (amqp_connection_close_t *)reply.reply.decoded;
    					sprintf(rtnmsg, "%s: server connection error %d, message: %.*s\n",
    							strModule.c_str(),
    							m->reply_code,
    							(int) m->reply_text.len, (char *) m->reply_text.bytes);
    				break;
    				}
    			case AMQP_CHANNEL_CLOSE_METHOD: 
    				{
    					amqp_channel_close_t *m = (amqp_channel_close_t *)reply.reply.decoded;
    					sprintf(rtnmsg, "%s: server channel error %d, message: %.*s\n",
    							strModule.c_str(),
    							m->reply_code,
    							(int) m->reply_text.len, (char *) m->reply_text.bytes);
    					break;
    				}
    			default:
    				sprintf(rtnmsg, "%s: unknown server error, method id 0x%08X\n", strModule, reply.reply.id);
    				break;
    		}
    		break;
    	}
    	ErrorReturn = rtnmsg;
    	return -1;
    }

还有一些常量的定义：

//RabbitMq服务端ip，端口，用户名，密码
    static const char connect_host[] = ""; 
    static const int connect_port = 8888;
    static const char connect_user[] = "rabbitmq"; 
    static const char connect_pwd[] = "rabbitmq";
    
    //频道号，队列名，交换机名，路由名
    static int channel_id = 1;
    static const char queue_name[] = "queue_name";
    static const char exchange_name[] = "exchange_name";
    static const char routing_keys_name[] = "test.1";

代码中的注释已经足够详细了，就不再多解释。在后续文章中，会介绍基于此封装类实现的RabbitMQ的各种模式的消息发布与消费。