RabbitMQ安裝：

一、RaibbitMQ伺服器安裝

1． 準備工作。如果之前安裝過RabbitMQ軟體，若想重新安裝，必須先把之前的RabbitMQ相關軟體解除安裝。

安裝具體過程：

1.右鍵，管理員身份執行：otp_win32_R16801.exe(不同版本可能命名字不一樣)，選擇next

 如果不是管理員身份執行，啟動Rabbitmq不能成功：Failed to start service RabbitMQ. Error: 拒絕訪問。

解決辦法：解除安裝，清理登錄檔重新安裝erlang，先解除安裝RabbitMQ和Erlang，再清理登錄檔。

清理的時候如果找不到登錄檔項，可以用登錄檔的搜尋功能搜尋：RabbitMQ、ErlSrv，將對應的項全部刪除。

2.預設安裝在C盤，建議程式安裝在非系統盤比如D盤（如果安裝在C盤可能會出現一些許可權問題），修改好安裝路徑後，選next：

3.進入安裝程式，選擇install，即可完成安裝：

3． 安裝RabbitMQ伺服器軟體。到這個頁面下載：

安裝具體過程：

1. 雙擊rabbitmq-server-3.1.1.exe。選擇next:

2.  預設安裝在C盤，直接安裝即可。

4． 還需要配置環境變數:

 4.1給path變數新增內容，在其後面增加：;%RABBITMQ_SERVER%\sbin （注意前面的分號），然後確定即可

 4.2 新增環境變數：RABBITMQ_SERVER 

環境變數RABBITMQ_SERVER 的值為：D:\My-Softwar-Installed\RabbitMQ Server\rabbitmq_server-3.1.3

現在開啟windows命令列（“cmd”），輸入rabbitmq-service

如果出現如下所示提示，即表示環境變數配置成功。

 安裝完成之後可以在系統的服務中看到RabbitMQ這個服務。

5． Rabbit還自帶監控功能. 
cmd進到sbin目錄，鍵入rabbitmq-plugins enable rabbitmq_management啟用監控管理，然後重啟Rabbitmq伺服器。 開啟網址http://localhost:55672

，使用者名稱和密碼都是guest。 

預設使用者名稱為guest,密碼：guest

如果沒有出現以上頁面，嘗試在windows命令列中輸入(以管理員方式執行)：

rabbitmq-plugins enable rabbitmq_management

然後執行下面的命令來安裝：

rabbitmq-service stop

rabbitmq-service install

rabbitmq-service start

二、 使用VS2010編譯RibbitMQ

3. 將 rabbitmq-codegen 中的內容拷貝到 rabbitmq-c-master 中的 codegen 目錄下（如果沒有該目錄請自行建立）.

4. 下載cmake並安裝。

按下一步直接安裝：

      這裡可以更改路徑（注意安裝路徑不要包含中文）,安裝完畢即可。這時桌面會生成一個cmake_gui的快捷方式。

5. 使用cmake編譯rabbitmq-c原始碼。

開啟cmake。

source處（紅圈處）填你的rabbitmq-c-master原始碼路徑，比如我放在d:/rabbitmq-c-master；下面的build處（黃圈處）填你source處的下面build資料夾（如果沒有build資料夾則建立一個）

點configure，這時會詢問選擇何種編譯器，選擇VS2010。

點finish。這時cmake會開始編譯，但是一會他可能會彈出如下錯誤資訊:

直接把ENABLE_SSL_SUPPORT去掉，再按configure即可。

等編譯完成，再點Generate按鈕。整個編譯過程即完成。

6. 開啟VS2010。用VS2010開啟rabbit-c-master/build下的rabbitmq-c.sln專案

7. 點選生成/生成解決方案（或按F7）

8. 專案此時會生成好：

9. 將D:\rabbitmq-c-master\build\librabbitmq\Debug下(注意D:\rabbitmq-c-master\build為前面cmake編譯生成檔案路徑，下同)的rabbitmq.1.dll動態連線庫拷到D:\rabbitmq-c-master\rabbitmq-c-master\build\examples\Debug資料夾下。

10. 現在所有的rabbit-c的example都可以運行了。至此，可以按照https://github.com/alanxz/rabbitmq-c 上的說明，執行測試了。

開啟“cmd”，進入到D:\rabbitmq-c-master\build\examples\Debug資料夾下：

輸入amqp_listen.exe localhost 5672 amq.direct test:

再在另一個cmd中也到達D:\rabbitmq-c-master\build\examples\Debug

輸入amqp_sendstring.exe localhost 5672 amq.direct test 

“hello world”

執行之後如果第一個終端出現如下介面，則說明整個rabbitMQ配置成功。

至此，說明Raabit-C客戶端與服務端都以成功。

RabbitMQ使用說明

一、 基本概念：

（一）基本概念

RabbitMQ是流行的開源訊息佇列系統，用erlang語言開發。我曾經對這門語言挺有興趣，學過一段時間，後來沒堅持。RabbitMQ是AMQP（高階訊息佇列協議）的標準實現。如果不熟悉AMQP，直接看RabbitMQ的文件會比較困難。不過它也只有幾個關鍵概念，這裡簡單介紹。

RabbitMQ的結構圖如下：

幾個概念說明：

Broker：簡單來說就是訊息佇列伺服器實體。
Exchange：訊息交換機，它指定訊息按什麼規則，路由到哪個佇列。
Queue：訊息佇列載體，每個訊息都會被投入到一個或多個佇列。
Binding：繫結，它的作用就是把exchange和queue按照路由規則繫結起來。
Routing Key：路由關鍵字，exchange根據這個關鍵字進行訊息投遞。
vhost：虛擬主機，一個broker裡可以開設多個vhost，用作不同使用者的許可權分離。
producer：訊息生產者，就是投遞訊息的程式。
consumer：訊息消費者，就是接受訊息的程式。
channel：訊息通道，在客戶端的每個連線裡，可建立多個channel，每個channel代表一個會話任務。

訊息佇列的使用過程大概如下：

（1）客戶端連線到訊息佇列伺服器，開啟一個channel。
（2）客戶端宣告一個exchange，並設定相關屬性。
（3）客戶端宣告一個queue，並設定相關屬性。
（4）客戶端使用routing key，在exchange和queue之間建立好繫結關係。
（5）客戶端投遞訊息到exchange。

exchange接收到訊息後，就根據訊息的key和已經設定的binding，進行訊息路由，將訊息投遞到一個或多個佇列裡。

exchange也有幾個型別，完全根據key進行投遞的叫做Direct交換機，例如，繫結時設定了routing key為”abc”，那麼客戶端提交的訊息，只有設定了key為”abc”的才會投遞到佇列。對key進行模式匹配後進行投遞的叫做Topic交換機，符號”#”匹配一個或多個詞，符號”*”匹配正好一個詞。例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”。還有一種不需要key的，叫做Fanout交換機，它採取廣播模式，一個訊息進來時，投遞到與該交換機繫結的所有佇列。

RabbitMQ支援訊息的持久化，也就是資料寫在磁碟上，為了資料安全考慮，我想大多數使用者都會選擇持久化。訊息佇列持久化包括3個部分：
（1）exchange持久化，在宣告時指定durable => 1
（2）queue持久化，在宣告時指定durable => 1
（3）訊息持久化，在投遞時指定delivery_mode => 2（1是非持久化）

如果exchange和queue都是持久化的，那麼它們之間的binding也是持久化的。如果exchange和queue兩者之間有一個持久化，一個非持久化，就不允許建立繫結。

RabbitMQ-C客戶端使用說明 rabbitmq-c是一個用於C語言的，與AMQP server進行互動的client庫，AMQP協議為版本0-9-1。rabbitmq-c與server進行互動前需要首先進行login操作，在操作後，可以根據AMQP協議規範，執行一系列操作。 這裡，根據專案需求，只進行部分介面說明，文後附demo的github地址。 介面描述： amqp_connection_state_t amqp_new_connection(void); 介面說明：宣告一個新的amqp connection int amqp_open_socket(char const *hostname, int portnumber); 介面說明：獲取socket.

引數說明：hostname RabbitMQ server所在主機

portnumber RabbitMQ server監聽埠 

void amqp_set_sockfd(amqp_connection_state_t state,int sockfd);

介面說明：將amqp connection和sockfd進行繫結

amqp_rpc_reply_t amqp_login(amqp_connection_state_t state, char const *vhost,int channel_max,int frame_max,int heartbeat,amqp_sasl_method_enum sasl_method, ...);

介面說明：用於登入RabbitMQ server，主要目的為了進行許可權管理；

引數說明：state amqp connection

vhost rabbit-mq的虛機主機，是rabbit-mq進行許可權管理的最小單位

channel_max 最大連結數，此處設成0即可

frame_max 和客戶端通訊時所允許的最大的frame size.預設值為131072，增大這個值有助於提高吞吐，降低這個值有利於降低時延

heartbeat 含義未知，預設值填0

sasl_method 用於SSL鑑權，預設值參考後文demo

amqp_channel_open_ok_t *amqp_channel_open(amqp_connection_state_t state, amqp_channel_t channel);

介面說明：用於關聯conn和channel

amqp_exchange_declare_ok_t *amqp_exchange_declare(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t exchange, amqp_bytes_t type, amqp_boolean_t passive, amqp_boolean_t durable, amqp_table_t arguments); 

介面說明：宣告declare

引數說明：state

channel

exchange

type "fanout" "direct" "topic"三選一

passive

curable

arguments

amqp_queue_declare_ok_t *amqp_queue_declare(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_boolean_t passive, amqp_boolean_t durable, amqp_boolean_t exclusive, amqp_boolean_t auto_delete, amqp_table_t arguments); 

介面說明：宣告queue

引數說明：state amqp connection

channel 

queue queue name

passive 

durable 佇列是否持久化

exclusive 當前連線不在時，佇列是否自動刪除

aoto_delete 沒有consumer時，佇列是否自動刪除

arguments 用於拓展引數，比如x-ha-policy用於mirrored queue

amqp_queue_bind_ok_t *amqp_queue_bind(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_bytes_t exchange, amqp_bytes_t routing_key, amqp_tab le_t arguments);

介面說明：宣告binding 

amqp_basic_qos_ok_t *amqp_basic_qos(amqp_connection_state_t state, amqp_channel_t channel, uint32_t prefetch_size, uint16_t prefetch_count, amqp_boolean_t global);

介面說明：qos是 quality of service，我們這裡使用主要用於控制預取訊息數，避免訊息按條數均勻分配，需要和no_ack配合使用

引數說明：state

channel

prefetch_size 以bytes為單位，0為unlimited

prefetch_count 預取的訊息條數

global

amqp_basic_consume_ok_t *amqp_basic_consume(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_bytes_t consumer_tag, amqp_boolean_t no_local, amqp_boolean_t no_ack, amqp_boolean_t exclusive, amqp_table_t arguments); 

介面說明：開始一個queue consumer

引數說明：state

channel

queue

consumer_tag

no_local

no_ack 是否需要確認訊息後再從佇列中刪除訊息

exclusive

arguments

int amqp_basic_ack(amqp_connection_state_t state,amqp_channel_t channel,uint64_t delivery_tag,amqp_boolean_t multiple);

int amqp_basic_publish(amqp_connection_state_t state,amqp_channel_t channel,amqp_bytes_t exchange,amqp_bytes_t routing_key,amqp_boolean_t mandatory,amqp_boolean_t immediate,struct amqp_basic_properties_t_ const *properties,amqp_bytes_t body);

介面說明：釋出訊息

引數說明：state 

channel

exchange 

routing_key 當exchange為預設“”時，此處填寫queue_name，當exchange為direct，此處為binding_key

mandatory 參見參考文獻2

immediate 同上

properties 更多屬性，如何設定訊息持久化，參見文後demo

body 訊息體

amqp_rpc_reply_t amqp_channel_close(amqp_connection_state_t state,amqp_channel_t channel,int code);

amqp_rpc_reply_t amqp_connection_close(amqp_connection_state_t state,int code);

int amqp_destroy_connection(amqp_connection_state_t state);

如何consume訊息，參見文後demo。

demo

RabbitMQ-C客戶端使用說明 rabbitmq-c是一個用於C語言的，與AMQP server進行互動的client庫，AMQP協議為版本0-9-1。rabbitmq-c與server進行互動前需要首先進行login操作，在操作後，可以根據AMQP協議規範，執行一系列操作。 這裡，根據專案需求，只進行部分介面說明，文後附demo的github地址。 介面描述： amqp_connection_state_t amqp_new_connection(void); 介面說明：宣告一個新的amqp connection int amqp_open_socket(char const *hostname, int portnumber); 介面說明：獲取socket. 引數說明：hostname RabbitMQ server所在主機 portnumber RabbitMQ server監聽埠  void amqp_set_sockfd(amqp_connection_state_t state,int sockfd); 介面說明：將amqp connection和sockfd進行繫結 amqp_rpc_reply_t amqp_login(amqp_connection_state_t state, char const *vhost,int channel_max,int frame_max,int heartbeat,amqp_sasl_method_enum sasl_method, ...); 介面說明：用於登入RabbitMQ server，主要目的為了進行許可權管理； 引數說明：state amqp connection vhost rabbit-mq的虛機主機，是rabbit-mq進行許可權管理的最小單位 channel_max 最大連結數，此處設成0即可 frame_max 和客戶端通訊時所允許的最大的frame size.預設值為131072，增大這個值有助於提高吞吐，降低這個值有利於降低時延 heartbeat 含義未知，預設值填0 sasl_method 用於SSL鑑權，預設值參考後文demo amqp_channel_open_ok_t *amqp_channel_open(amqp_connection_state_t state, amqp_channel_t channel); 介面說明：用於關聯conn和channel amqp_exchange_declare_ok_t *amqp_exchange_declare(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t exchange, amqp_bytes_t type, amqp_boolean_t passive, amqp_boolean_t durable, amqp_table_t arguments);  介面說明：宣告declare 引數說明：state channel exchange type "fanout" "direct" "topic"三選一 passive curable arguments amqp_queue_declare_ok_t *amqp_queue_declare(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_boolean_t passive, amqp_boolean_t durable, amqp_boolean_t exclusive, amqp_boolean_t auto_delete, amqp_table_t arguments);  介面說明：宣告queue 引數說明：state amqp connection channel  queue queue name passive  durable 佇列是否持久化 exclusive 當前連線不在時，佇列是否自動刪除 aoto_delete 沒有consumer時，佇列是否自動刪除 arguments 用於拓展引數，比如x-ha-policy用於mirrored queue amqp_queue_bind_ok_t *amqp_queue_bind(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_bytes_t exchange, amqp_bytes_t routing_key, amqp_tab le_t arguments); 介面說明：宣告binding  amqp_basic_qos_ok_t *amqp_basic_qos(amqp_connection_state_t state, amqp_channel_t channel, uint32_t prefetch_size, uint16_t prefetch_count, amqp_boolean_t global); 介面說明：qos是 quality of service，我們這裡使用主要用於控制預取訊息數，避免訊息按條數均勻分配，需要和no_ack配合使用 引數說明：state channel prefetch_size 以bytes為單位，0為unlimited prefetch_count 預取的訊息條數 global amqp_basic_consume_ok_t *amqp_basic_consume(amqp_connection_state_t state, amqp_channel_t channel, amqp_bytes_t queue, amqp_bytes_t consumer_tag, amqp_boolean_t no_local, amqp_boolean_t no_ack, amqp_boolean_t exclusive, amqp_table_t arguments);  介面說明：開始一個queue consumer 引數說明：state channel queue consumer_tag no_local no_ack 是否需要確認訊息後再從佇列中刪除訊息 exclusive arguments int amqp_basic_ack(amqp_connection_state_t state,amqp_channel_t channel,uint64_t delivery_tag,amqp_boolean_t multiple); int amqp_basic_publish(amqp_connection_state_t state,amqp_channel_t channel,amqp_bytes_t exchange,amqp_bytes_t routing_key,amqp_boolean_t mandatory,amqp_boolean_t immediate,struct amqp_basic_properties_t_ const *properties,amqp_bytes_t body); 介面說明：釋出訊息 引數說明：state  channel exchange  routing_key 當exchange為預設“”時，此處填寫queue_name，當exchange為direct，此處為binding_key mandatory 參見參考文獻2 immediate 同上 properties 更多屬性，如何設定訊息持久化，參見文後demo body 訊息體 amqp_rpc_reply_t amqp_channel_close(amqp_connection_state_t state,amqp_channel_t channel,int code); amqp_rpc_reply_t amqp_connection_close(amqp_connection_state_t state,int code); int amqp_destroy_connection(amqp_connection_state_t state); 如何consume訊息，參見文後demo。 demo 編譯需要librabbitmq.a庫，同時需要rabbitmq-c提供的幾個標頭檔案（amqp.h和amqp_framing.h）以及utils.c檔案，這些在github project頁面均可獲得。 參考文獻

nt main(int argc, const char **argv) {

const char *hostname = "localhost";  // 主機名

int port = 5672;   // 埠

const char *exchange = "";   //交換機為預設時設為空字串

const char *queuename = "QueueName1";   // 佇列名

amqp_socket_t *socket = NULL;

int sockfd;

int channelid = 1;   // 預設開啟的頻道

amqp_connection_state_t conn;

conn = amqp_new_connection();  // 新建一個連線

socket = amqp_tcp_socket_new(conn);  // 新建一個套接字

die_on_error(sockfd = amqp_socket_open(socket,hostname, port), "Opening socket");  //開啟套接字

die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),"Logging in");   // 登入伺服器

amqp_channel_open(conn, channelid);  // 開啟一個頻道

die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");   // ampq_get_rpc_reply 是一個清除某些全域性變數的函式

amqp_queue_declare(conn,channelid,amqp_cstring_bytes(queuename),0,1,0,0,amqp_empty_table);  // 宣告一個佇列,注意對照API看引數資訊

amqp_basic_qos(conn,channelid,0,1,0);  //基本服務處理，可實現預先處理多少個資訊

amqp_basic_consume(conn,channelid,amqp_cstring_bytes(queuename),amqp_empty_bytes,0,1,0,amqp_empty_table);   // 開始一個佇列消費

die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");

{

相關推薦