目錄

?

?

概念及注意

代碼與實(shí)例

---

?

概念及注意

這里C接口中有2個(gè)函數(shù)一個(gè)是：

amqp_basic_get()

另外一個(gè)是：

amqp_basic_consume()

前者可以一條一條的讀取，后者是一次讀取一大坨數(shù)據(jù)。所以目前要使用

amqp_basic_get()去讀取RabbitMQ的數(shù)據(jù)，同時(shí)使用amqp_read_message()把數(shù)據(jù)給讀出來(lái)。

讀完一條數(shù)據(jù)就發(fā)一個(gè)確認(rèn)，這樣的話即能者多勞。

?

代碼與實(shí)例

程序運(yùn)行截圖如下：

消息隊(duì)列端：

源碼如下：

consumer端：

#define _CRT_SECURE_NO_WARNINGS #include <iostream> #include <string> #include <amqp.h> #include <time.h> #include <windows.h>#include <amqp_framing.h> #include <amqp_tcp_socket.h> using namespace std;void delay_msec(int msec) { clock_t now = clock();while(clock() - now < msec); }int main(int *argc, int *argv[]){string name = (char *)argv[1];string delayStr = (char*)argv[2];//cout << "The delayStr is : " << delayStr << endl;int delay = std::stoi(delayStr);//cout << "The delay is : " << delay << endl;//string name = "one";//int delay = 1;string hostName = "127.0.0.1";int port = 5672;amqp_socket_t *socket = nullptr;amqp_connection_state_t conn;conn = amqp_new_connection();socket = amqp_tcp_socket_new(conn);if(!socket){cout << "create socket failed!";exit(1);}if(amqp_socket_open(socket, hostName.c_str(), port)){cout << "opening TCP socket failed" << endl;exit(1);}//登錄if(1 != amqp_login(conn, "/vhost_cff", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "cff", "123").reply_type){cout << "login failed" << endl;exit(1);}amqp_channel_open(conn, 1);while(1){amqp_basic_get(conn, 1, amqp_cstring_bytes("test_work_queue"), 1);amqp_message_t *msg = new amqp_message_t;amqp_read_message(conn, 1, msg, 0);cout << "【" << name << "】 The result is : " << (char *)msg->body.bytes << endl;amqp_destroy_message(msg);delete msg;delay_msec(1000 * delay);}amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);amqp_connection_close(conn, AMQP_REPLY_SUCCESS);getchar();return 0; }

producer端：

#define _CRT_SECURE_NO_WARNINGS #include <iostream> #include <string> #include <time.h> #include <amqp.h> #include <windows.h> #include <amqp_framing.h> #include <amqp_tcp_socket.h> using namespace std;void delay_msec(int msec) { clock_t now = clock();while(clock() - now < msec); }int main(int *argc, int *argv[]){string hostName = "127.0.0.1";int port = 5672;amqp_socket_t *socket = nullptr;amqp_connection_state_t conn;conn = amqp_new_connection();socket = amqp_tcp_socket_new(conn);if(!socket){cout << "create socket failed!";exit(1);}if(amqp_socket_open(socket, hostName.c_str(), port)){cout << "opening TCP socket failed" << endl;exit(1);}//登錄if(1 != amqp_login(conn, "/vhost_cff", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "cff", "123").reply_type){cout << "login failed" << endl;}//管道amqp_channel_open(conn, 1);int i = 0;while(1){string str = "Hello " + to_string(i);char message[64] = {'\0'};strcpy(message, str.c_str());amqp_bytes_t message_bytes;message_bytes.len = sizeof(message);message_bytes.bytes = message;amqp_basic_publish(conn, 1, amqp_cstring_bytes(""), amqp_cstring_bytes("test_work_queue"), 0, 0, nullptr, message_bytes);cout << "send msg over!" << str << endl;i++;delay_msec(100);if(i == 10000)break;}amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);amqp_connection_close(conn, AMQP_REPLY_SUCCESS);getchar();return 0; }

?

源碼下載地址：

https://github.com/fengfanchen/CAndCPP/tree/master/RabbitMQFair

總結(jié)

以上是生活随笔為你收集整理的RabbitMQ笔记-使用rabbitmq-c实现Fair dispatch(公平分发)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。