Rabbitmq C++客戶端（基於rabbitmq-c）
阿新 • • 發佈：2019-01-10
最近專案使用了rabbitmq訊息佇列，需要基於rabbitmq-c(https://github.com/alanxz/rabbitmq-c)進行一層封裝，然後根據官方自帶示例以及網上程式碼整理了一下。
傳送者需要宣告exchange，並指定routekey進行資料傳送；消費者需要宣告一個Queue並繫結exchange和routekey，進行資料接收。
RabbitmqClient標頭檔案：
#ifndef RABBITMQ_CLIENT_H_ #define RABBITMQ_CLIENT_H_ #include <string> #include <vector> #include "amqp_tcp_socket.h" using std::string; using std::vector; class CRabbitmqClient{ public: CRabbitmqClient(); ~CRabbitmqClient(); int Connect(const string &strHostname, int iPort, const string &strUser, const string &strPasswd); int Disconnect(); /** * @brief ExchangeDeclare 宣告exchange * @param [in] strExchange * @param [in] strType * @return 等於0值代表成功建立exchange，小於0代表錯誤 */ int ExchangeDeclare(const string &strExchange, const string &strType); /** * @brief QueueDeclare 宣告訊息佇列 * @param [in] strQueueName 訊息佇列例項 * @param * @return 等於0值代表成功建立queue，小於0代表錯誤 */ int QueueDeclare(const string &strQueueName); /** * @brief QueueBind 將佇列，交換機和繫結規則繫結起來形成一個路由表 * @param [in] strQueueName 訊息佇列 * @param [in] strExchange 交換機名稱 * @param [in] strBindKey 路由名稱 “msg.#” “msg.weather.**” * @return 等於0值代表成功繫結，小於0代表錯誤 */ int QueueBind(const string &strQueueName, const string &strExchange, const string &strBindKey); /** * @brief QueueUnbind 將佇列，交換機和繫結規則繫結解除 * @param [in] strQueueName 訊息佇列 * @param [in] strExchange 交換機名稱 * @param [in] strBindKey 路由名稱 “msg.#” “msg.weather.**” * @return 等於0值代表成功繫結，小於0代表錯誤 */ int QueueUnbind(const string &strQueueName, const string &strExchange, const string &strBindKey); /** * @brief QueueDelete 刪除訊息佇列。 * @param [in] strQueueName 訊息佇列名稱 * @param [in] iIfUnused 訊息佇列是否在用，1 則論是否在用都刪除 * @return 等於0值代表成功刪除queue，小於0代表錯誤 */ int QueueDelete(const string &strQueueName, int iIfUnused); /** * @brief Publish 釋出訊息 * @param [in] strMessage 訊息實體 * @param [in] strExchange 交換器 * @param [in] strRoutekey 路由規則 * 1.Direct Exchange – 處理路由鍵。需要將一個佇列繫結到交換機上，要求該訊息與一個特定的路由鍵完全匹配。 * 2.Fanout Exchange – 不處理路由鍵。將佇列繫結到交換機上。一個傳送到交換機的訊息都會被轉發到與該交換機繫結的所有佇列上。 * 3.Topic Exchange – 將路由鍵和某模式進行匹配。此時佇列需要繫結要一個模式上。符號“#”匹配一個或多個詞，符號“*”匹配不多不少一個詞。 * 因此“audit.#”能夠匹配到“audit.irs.corporate”，但是“audit.*” 只會匹配到“audit.irs” * @return 等於0值代表成功傳送訊息實體，小於0代表傳送錯誤 */ int Publish(const string &strMessage, const string &strExchange, const string &strRoutekey); /** * @brief consumer 消費訊息 * @param [in] strQueueName 佇列名稱 * @param [out] message_array 獲取的訊息實體 * @param [int] GetNum 需要取得的訊息個數 * @param [int] timeout 取得的訊息是延遲，若為NULL，表示持續取，無延遲，阻塞狀態 * @return 等於0值代表成功，小於0代表錯誤，錯誤資訊從ErrorReturn返回 */ int Consumer(const string &strQueueName, vector<string> &message_array, int GetNum = 1, struct timeval *timeout = NULL); private: CRabbitmqClient(const CRabbitmqClient & rh); void operator=(const CRabbitmqClient & rh); int ErrorMsg(amqp_rpc_reply_t x, char const *context); string m_strHostname; // amqp主機 int m_iPort; // amqp埠 string m_strUser; string m_strPasswd; int m_iChannel; amqp_socket_t *m_pSock; amqp_connection_state_t m_pConn; }; #endif
RabbitmqClient原始檔：
#include "RabbitmqClient.h" #include <unistd.h> CRabbitmqClient::CRabbitmqClient() : m_strHostname("") , m_iPort(0) , m_strUser("") , m_strPasswd("") , m_iChannel(1) //預設用1號通道，通道無所謂 , m_pSock(NULL) , m_pConn(NULL) { } CRabbitmqClient::~CRabbitmqClient() { if (NULL != m_pConn) { Disconnect(); m_pConn = NULL; } } int CRabbitmqClient::Connect(const string &strHostname, int iPort, const string &strUser, const string &strPasswd) { m_strHostname = strHostname; m_iPort = iPort; m_strUser = strUser; m_strPasswd = strPasswd; m_pConn = amqp_new_connection(); if (NULL == m_pConn) { fprintf(stderr, "amqp new connection failed\n"); return -1; } m_pSock = amqp_tcp_socket_new(m_pConn); if (NULL == m_pSock) { fprintf(stderr, "amqp tcp new socket failed\n"); return -2; } int status = amqp_socket_open(m_pSock, m_strHostname.c_str(), m_iPort); if (status<0) { fprintf(stderr, "amqp socket open failed\n"); return -3; } // amqp_login(amqp_connection_state_t state,char const *vhost, int channel_max, int frame_max, int heartbeat, amqp_sasl_method_enum sasl_method, ..) if (0 != ErrorMsg(amqp_login(m_pConn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, m_strUser.c_str(), m_strPasswd.c_str()), "Logging in")) { return -4; } return 0; } int CRabbitmqClient::Disconnect() { if (NULL != m_pConn) { if (0 != ErrorMsg(amqp_connection_close(m_pConn, AMQP_REPLY_SUCCESS), "Closing connection")) return -1; if (amqp_destroy_connection(m_pConn) < 0) return -2; m_pConn = NULL; } return 0; } int CRabbitmqClient::ExchangeDeclare(const string &strExchange, const string &strType) { amqp_channel_open(m_pConn, m_iChannel); amqp_bytes_t _exchange = amqp_cstring_bytes(strExchange.c_str()); amqp_bytes_t _type = amqp_cstring_bytes(strType.c_str()); int _passive= 0; int _durable= 0; // 交換機是否持久化 amqp_exchange_declare(m_pConn, m_iChannel, _exchange, _type, _passive, _durable, 0, 0, amqp_empty_table); if (0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "exchange_declare")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -1; } amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return 0; } int CRabbitmqClient::QueueDeclare(const string &strQueueName) { if(NULL == m_pConn) { fprintf(stderr, "QueueDeclare m_pConn is null\n"); return -1; } amqp_channel_open(m_pConn, m_iChannel); amqp_bytes_t _queue = amqp_cstring_bytes(strQueueName.c_str()); int32_t _passive = 0; int32_t _durable = 0; int32_t _exclusive = 0; int32_t _auto_delete = 1; amqp_queue_declare(m_pConn, m_iChannel, _queue, _passive, _durable, _exclusive, _auto_delete, amqp_empty_table); if (0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "queue_declare")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -1; } amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return 0; } int CRabbitmqClient::QueueBind(const string &strQueueName, const string &strExchange, const string &strBindKey) { if(NULL == m_pConn) { fprintf(stderr, "QueueBind m_pConn is null\n"); return -1; } amqp_channel_open(m_pConn, m_iChannel); amqp_bytes_t _queue = amqp_cstring_bytes(strQueueName.c_str()); amqp_bytes_t _exchange = amqp_cstring_bytes(strExchange.c_str()); amqp_bytes_t _routkey = amqp_cstring_bytes(strBindKey.c_str()); amqp_queue_bind(m_pConn, m_iChannel, _queue, _exchange, _routkey, amqp_empty_table); if(0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "queue_bind")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -1; } amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return 0; } int CRabbitmqClient::QueueUnbind(const string &strQueueName, const string &strExchange, const string &strBindKey) { if(NULL == m_pConn) { fprintf(stderr, "QueueUnbind m_pConn is null\n"); return -1; } amqp_channel_open(m_pConn, m_iChannel); amqp_bytes_t _queue = amqp_cstring_bytes(strQueueName.c_str()); amqp_bytes_t _exchange = amqp_cstring_bytes(strExchange.c_str()); amqp_bytes_t _routkey = amqp_cstring_bytes(strBindKey.c_str()); amqp_queue_unbind(m_pConn, m_iChannel, _queue, _exchange, _routkey, amqp_empty_table); if(0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "queue_unbind")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -1; } amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return 0; } int CRabbitmqClient::QueueDelete(const string &strQueueName, int iIfUnused) { if(NULL == m_pConn) { fprintf(stderr, "QueueDelete m_pConn is null\n"); return -1; } amqp_channel_open(m_pConn, m_iChannel); if(0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "open channel")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -2; } amqp_queue_delete(m_pConn, m_iChannel, amqp_cstring_bytes(strQueueName.c_str()), iIfUnused, 0); if(0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "delete queue")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -3; } amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return 0; } int CRabbitmqClient::Publish(const string &strMessage, const string &strExchange, const string &strRoutekey) { if (NULL == m_pConn) { fprintf(stderr, "publish m_pConn is null, publish failed\n"); return -1; } amqp_channel_open(m_pConn, m_iChannel); if(0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "open channel")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -2; } amqp_bytes_t message_bytes; message_bytes.len = strMessage.length(); message_bytes.bytes = (void *)(strMessage.c_str()); //fprintf(stderr, "publish message(%d): %.*s\n", (int)message_bytes.len, (int)message_bytes.len, (char *)message_bytes.bytes); /* amqp_basic_properties_t props; props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG; props.content_type = amqp_cstring_bytes(m_type.c_str()); props.delivery_mode = m_durable; // persistent delivery mode */ amqp_bytes_t exchange = amqp_cstring_bytes(strExchange.c_str()); amqp_bytes_t routekey = amqp_cstring_bytes(strRoutekey.c_str()); //if (0 != amqp_basic_publish(m_pConn, m_iChannel, exchange, routekey, 0, 0, &props, message_bytes)) { if (0 != amqp_basic_publish(m_pConn, m_iChannel, exchange, routekey, 0, 0, NULL, message_bytes)) { fprintf(stderr, "publish amqp_basic_publish failed\n"); if (0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "amqp_basic_publish")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -3; } } amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return 0; } int CRabbitmqClient::Consumer(const string &strQueueName, vector<string> &message_array, int GetNum, struct timeval *timeout) { if (NULL == m_pConn) { fprintf(stderr, "Consumer m_pConn is null, Consumer failed\n"); return -1; } amqp_channel_open(m_pConn, m_iChannel); if (0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "open channel")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -2; } amqp_basic_qos(m_pConn, m_iChannel, 0, GetNum, 0); int ack = 1; // no_ack 是否需要確認訊息後再從佇列中刪除訊息 amqp_bytes_t queuename= amqp_cstring_bytes(strQueueName.c_str()); amqp_basic_consume(m_pConn, m_iChannel, queuename, amqp_empty_bytes, 0, ack, 0, amqp_empty_table); if (0 != ErrorMsg(amqp_get_rpc_reply(m_pConn), "Consuming")) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -3; } int hasget = 0; amqp_rpc_reply_t res; amqp_envelope_t envelope; while (GetNum > 0) { amqp_maybe_release_buffers(m_pConn); res = amqp_consume_message(m_pConn, &envelope, timeout, 0); if (AMQP_RESPONSE_NORMAL != res.reply_type) { fprintf(stderr, "Consumer amqp_channel_close failed\n"); amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); if (0 == hasget) return -res.reply_type; else return 0; } string str((char *)envelope.message.body.bytes, (char *)envelope.message.body.bytes + envelope.message.body.len); message_array.push_back(str); int rtn = amqp_basic_ack(m_pConn, m_iChannel, envelope.delivery_tag, 1); amqp_destroy_envelope(&envelope); if (rtn != 0) { amqp_channel_close(m_pConn, m_iChannel, AMQP_REPLY_SUCCESS); return -4; } GetNum--; hasget++; usleep(1); } return 0; } int CRabbitmqClient::ErrorMsg(amqp_rpc_reply_t x, char const *context) { switch (x.reply_type) { case AMQP_RESPONSE_NORMAL: return 0; case AMQP_RESPONSE_NONE: fprintf(stderr, "%s: missing RPC reply type!\n", context); break; case AMQP_RESPONSE_LIBRARY_EXCEPTION: fprintf(stderr, "%s: %s\n", context, amqp_error_string2(x.library_error)); break; case AMQP_RESPONSE_SERVER_EXCEPTION: switch (x.reply.id) { case AMQP_CONNECTION_CLOSE_METHOD: { amqp_connection_close_t *m = (amqp_connection_close_t *)x.reply.decoded; fprintf(stderr, "%s: server connection error %uh, message: %.*s\n", context, 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 *)x.reply.decoded; fprintf(stderr, "%s: server channel error %uh, message: %.*s\n", context, m->reply_code, (int)m->reply_text.len, (char *)m->reply_text.bytes); break; } default: fprintf(stderr, "%s: unknown server error, method id 0x%08X\n", context, x.reply.id); break; } break; } return -1; }
main檔案：
#include <stdio.h> #include <stdlib.h> #include <string.h> #include <unistd.h> #include <iostream> #include "RabbitmqClient.h" int main() { CRabbitmqClient objRabbitmq; std::string strIP = "192.168.240.121"; int iPort = 8856; std::string strUser = "rabbitadmin"; std::string strPasswd = "123321"; int iRet = objRabbitmq.Connect(strIP, iPort, strUser, strPasswd); printf("Rabbitmq Connect Ret: %d\n", iRet); std::string strExchange = "ExchangeTest"; std::string strRoutekey = "routekeyTest"; std::string strQueuename = "queueTest"; // 可選操作 Declare Exchange iRet = objRabbitmq.ExchangeDeclare(strExchange, "direct"); printf("Rabbitmq ExchangeDeclare Ret: %d\n", iRet); // 可選操作（接收） Declare Queue iRet = objRabbitmq.QueueDeclare(strQueuename); printf("Rabbitmq QueueDeclare Ret: %d\n", iRet); // 可選操作（接收） Queue Bind iRet = objRabbitmq.QueueBind(strQueuename, strExchange, strRoutekey); printf("Rabbitmq QueueBind Ret: %d\n", iRet); // Send Msg std::string strSendMsg1 = "rabbitmq send test msg1"; std::string strSendMsg2 = "rabbitmq send test msg2"; iRet = objRabbitmq.Publish(strSendMsg1, strExchange, strRoutekey); printf("Rabbitmq Publish 1 Ret: %d\n", iRet); iRet = objRabbitmq.Publish(strSendMsg2, strExchange, strRoutekey); printf("Rabbitmq Publish 2 Ret: %d\n", iRet); // Recv Msg std::vector<std::string> vecRecvMsg; iRet = objRabbitmq.Consumer(strQueuename, vecRecvMsg, 2); printf("Rabbitmq Consumer Ret: %d\n", iRet); for (size_t i=0; i<vecRecvMsg.size(); ++i) { printf("Consumer: %s\n", vecRecvMsg[i].c_str()); } objRabbitmq.Disconnect(); return 0; }
makefile檔案：
CC=gcc
CXX=g++
INC_PATH= ./
O_FLAG = -O0
CFLAGS += ${O_FLAG} -L $(PCAP_CFLAGS) -Wno-deprecated -Wall
LDFLAGS = -L $(PCAPLIB) $(LIBLINEAR) -L/usr/lib -lpthread -lrabbitmq
CFLAGS += -I$(INC_PATH) $(INCLUDE) -g -std=c++11
# 輸出檔名
TARGET= ./bin/Test
OUTPUT_PATH = ./obj
#設定VPATH 包含原始碼的子目錄列表
#新增原始檔
SUBINC = .
#新增標頭檔案
SUBDIR = .
#設定VPATH
INCLUDE = $(foreach n, $(SUBINC), -I$(INC_PATH)/$(n))
SPACE =
VPATH = $(subst $(SPACE),, $(strip $(foreach n,$(SUBDIR), $(INC_PATH)/$(n)))) $(OUTPUT_PATH)
C_SOURCES = $(notdir $(foreach n, $(SUBDIR), $(wildcard $(INC_PATH)/$(n)/*.c)))
CPP_SOURCES = $(notdir $(foreach n, $(SUBDIR), $(wildcard $(INC_PATH)/$(n)/*.cpp)))
C_OBJECTS = $(patsubst %.c, $(OUTPUT_PATH)/%.o, $(C_SOURCES))
CPP_OBJECTS = $(patsubst %.cpp, $(OUTPUT_PATH)/%.o, $(CPP_SOURCES))
CXX_SOURCES = $(CPP_SOURCES) $(C_SOURCES)
CXX_OBJECTS = $(CPP_OBJECTS) $(C_OBJECTS)
$(TARGET):$(CXX_OBJECTS)
$(CXX) -o [email protected] $(foreach n, $(CXX_OBJECTS), $(n)) $(foreach n, $(OBJS), $(n)) $(LDFLAGS)
#******************************************************************************#
# Bulid successful ! #
#******************************************************************************#
$(OUTPUT_PATH)/%.o:%.cpp
$(CXX) $< -c $(CFLAGS) -o [email protected]
$(OUTPUT_PATH)/%.o:%.c
$(CC) $< -c $(CFLAGS) -o [email protected]
mkdir:
mkdir -p $(dir $(TARGET))
mkdir -p $(OUTPUT_PATH)
rmdir:
rm -rf $(dir $(TARGET))
rm -rf $(OUTPUT_PATH)
clean:
rm -f $(OUTPUT_PATH)/*
rm -rf $(TARGET)程式在CentOS6.5版本下編譯，機器需要安裝rabbitmq-c庫。
程式執行輸出結果：
./bin/Test
Rabbitmq Connect Ret: 0
Rabbitmq ExchangeDeclare Ret: 0
Rabbitmq QueueDeclare Ret: 0
Rabbitmq QueueBind Ret: 0
Rabbitmq Publish 1 Ret: 0
Rabbitmq Publish 2 Ret: 0
Rabbitmq Consumer Ret: 0
Consumer: rabbitmq send test msg1
Consumer: rabbitmq send test msg2
Rabbitmq頁面顯示：
