C#教程之C#教程之RabbitMQ基礎入門篇

作者：warren

來源：http://wuchen1314.cnblogs.com/

聲明：原創博客請在轉載時保留原文鏈接或者在文章開頭加上本人博客地址，如發現錯誤，歡迎批評指正。凡是轉載於本人的文章，不能設置打賞功能，如有特殊需求請與本人聯系！

下載安裝

Erlang
RabbitMQ

啟動RabbitMQ管理平台插件

DOS下進入到安裝目錄\sbin,執行以下命令

rabbitmq-plugins enable rabbitmq_management 

當出現以下結果時，重啟RabbitMQ服務

set 3 plugins. Offline change; changes will take effect at broker restart.

訪問http://localhost:15672（賬號密碼：guest）

注意：以下為C#代碼，請引用NuGet包：RabbitMQ.Client

回到頂部

參考文章

RabbitMQ快速入門

回到頂部

名詞解析

P（Publisher）：生產者
C（Consumer）：消費者
Channel：信道
Queue：隊列
Exchange：信息交換機

回到頂部

簡單演示

信息發送端

static void Send() { //1. 實例化連接工廠 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立連接 using (var connection = factory.CreateConnection()) { //3. 創建信道 using (var channel = connection.CreateModel()) { //4. 聲明隊列 channel.QueueDeclare(queue: "rabbitmq", durable: false, exclusive: false, autoDelete: false, arguments: null); //5. 構建字節數據包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 發送數據包 channel.BasicPublish(exchange: "", routingKey: "rabbitmq", basicProperties: null, body: body); Console.WriteLine(" [x] Sent {0}", message); } } }

信息接收端

static void Receive() { //1. 實例化連接工廠 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立連接 using (var connection = factory.CreateConnection()) { //3. 創建信道 using (var channel = connection.CreateModel()) { //4. 聲明隊列 channel.QueueDeclare(queue: "rabbitmq", durable: false, exclusive: false, autoDelete: false, arguments: null); //5. 構造消費者實例 var consumer = new EventingBasicConsumer(channel); //6. 綁定消息接收后的事件委托 consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine(" [x] Received {0}", message); }; //7. 啟動消費者 channel.BasicConsume(queue: "rabbitmq", autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } }

回到頂部

輪詢調度

P生產的多個任務進入到隊列中，多個C間可以並行處理任務。默認情況下，RabbitMQ把信息按順序發送給每一個C。平均每個C將獲得同等數量的信息。

回到頂部

信息確認

按照最簡單的演示來說，信息一旦發送到C中，則該信息就會從隊列中移除。一旦中間信息處理異常/失敗，C端程序退出等，都將會導致信息未處理完成，而此時隊列中已將信息移除了，那么就會導致一系列的問題。我們可以在C端設置手動確認信息，從而解決上述問題的發生。
Receive代碼塊

//6. 綁定消息接收后的事件委托 consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine(" [x] Received {0}", message); Thread.Sleep(5000);//模擬耗時 Console.WriteLine(" [x] Done"); // 發送信息確認信號（手動信息確認） channel.BasicAck(ea.DeliveryTag, false); }; //7. 啟動消費者 /* autoAck參數屬性 true：自動信息確認，當C端接收到信息后，自動發送ack信號，不管信息是否處理完畢 false：關閉自動信息確認，通過調用BasicAck方法手動進行信息確認 */ channel.BasicConsume(queue: "rabbitmq", autoAck: false, consumer: consumer); 

回到頂部

信息持久化

當RabbitMQ退出或死機時會清空隊列和信息。通過將隊列和信息標記為持久的，來告知RabbitMQ將信息持久化。

Send代碼塊

//4. 聲明隊列 //durable設置為true，表示此隊列為持久的。 //注意：RabbitMQ不允許你使用不同的參數重新定義一個已經存在的隊列，所以你需要重啟服務/更改隊列名稱 channel.QueueDeclare(queue: "rabbitmq", durable: true, //標記隊列持久 exclusive: false, autoDelete: false, arguments: null); //設置IbasicProperties.SetPersistent屬性值為true來標記我們的消息持久化 var properties = channel.CreateBasicProperties(); properties.Persistent = true; //5. 構建字節數據包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 發送數據包 channel.BasicPublish(exchange: "", routingKey: "rabbitmq", basicProperties: properties, //指定BasicProperties body: body);

回到頂部

公平調度

上述演示中，如果隊列中存在多個信息，在開啟多個C的情況下，只有一個C忙個不停，另外的卻一直處於空閑狀態。通過調用BasicQos，告知RabbitMQ在某個C信息處理完畢，並且已經收到信息確認之后，才可以繼續發送信息到這個C。否則，將會把信息分發到另外空閑的C。

Receive代碼塊

//4. 聲明隊列 channel.QueueDeclare(queue: "rabbitmq",  durable: true,  exclusive: false,  autoDelete: false,  arguments: null); ////設置prefetchCount為1來告知RabbitMQ在未收到消費端的消息確認時，不再分發消息 channel.BasicQos(prefetchSize: 0,  prefetchCount: 1,  global: false);

回到頂部

發布/訂閱

上述中的演示，P推送信息至隊列中，C從隊列中處理信息。但是如果需要將P推送的信息至每個訂閱的C中處理信息，那么我們就可以使用Exchange。

fanout（將信息分發到exchange上綁定的所有隊列上）

Send代碼塊

//1. 實例化連接工廠 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立連接 using (var connection = factory.CreateConnection()) { //3. 創建信道 using (var channel = connection.CreateModel()) { //4. 聲明信息交換機 channel.ExchangeDeclare(exchange: "fanoutDemo", type: "fanout"); //5. 構建字節數據包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 發布到指定exchange，fanout類型的會忽視routingKey的值，所以無需填寫 channel.BasicPublish(exchange: "fanoutDemo", routingKey: "", basicProperties: null, body: body); Console.WriteLine(" [x] Sent {0}", message); } }

Receive代碼塊

//1. 實例化連接工廠 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立連接 using (var connection = factory.CreateConnection()) { //3. 創建信道 using (var channel = connection.CreateModel()) { //4. 聲明信息交換機 channel.ExchangeDeclare(exchange: "fanoutDemo", type: "fanout"); //生成隨機隊列名稱 var queueName = channel.QueueDeclare().QueueName; //綁定隊列到指定fanout類型exchange channel.QueueBind(queue: queueName, exchange: "fanoutDemo", routingKey: ""); //5. 構造消費者實例 var consumer = new EventingBasicConsumer(channel); //6. 綁定消息接收后的事件委托 consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine(" [x] Received {0}", message); }; channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } }

direct（C綁定的隊列名稱須和P發布指定的路由名稱一致）

Send代碼塊

//4. 聲明信息交換機 channel.ExchangeDeclare(exchange: "directDemo", type: "direct"); //5. 構建字節數據包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 發布到指定exchange channel.BasicPublish(exchange: "directDemo", routingKey: "a", basicProperties: null, body: body);

Receive代碼塊

//4. 聲明信息交換機 channel.ExchangeDeclare(exchange: "directDemo", type: "direct"); //生成隨機隊列名稱 var queueName = channel.QueueDeclare().QueueName; //綁定隊列到指定direct類型exchange channel.QueueBind(queue: queueName, exchange: "directDemo", routingKey: "b");

topic（支持通配符的路由規則）

通配字符：

• *：匹配一個單詞
• #：匹配0個或多個單詞
• .：僅作為分隔符

Send代碼塊

//4. 聲明信息交換機 channel.ExchangeDeclare(exchange: "topicDemo", type: "topic"); //5. 構建字節數據包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 發布到指定exchange channel.BasicPublish(exchange: "topicDemo", routingKey: "admin.user.error", //模擬后台用戶錯誤 basicProperties: null, body: body);

Receive代碼塊

//4. 聲明信息交換機 channel.ExchangeDeclare(exchange: "topicDemo", type: "topic"); //生成隨機隊列名稱 var queueName = channel.QueueDeclare().QueueName; //綁定隊列到指定topic類型exchange channel.QueueBind(queue: queueName, exchange: "topicDemo", routingKey: "admin.*.#"); //訂閱所有后台異常錯誤

回到頂部

RPC（遠程過程調用）

1. 進行遠程調用的客戶端需要指定接收遠程回調的隊列，並申明消費者監聽此隊列。
2. 遠程調用的服務端除了要申明消費端接收遠程調用請求外，還要將結果發送到客戶端用來監聽的結果的隊列中去。

客戶端代碼塊

var factory = new ConnectionFactory() { HostName = "localhost" }; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { var correlationId = Guid.NewGuid().ToString(); var replyQueue = channel.QueueDeclare().QueueName; var properties = channel.CreateBasicProperties(); properties.ReplyTo = replyQueue; properties.CorrelationId = correlationId; string number = args.Length > 0 ? args[0] : "30"; var body = Encoding.UTF8.GetBytes(number); //發布消息 channel.BasicPublish(exchange: "", routingKey: "rpc_queue", basicProperties: properties, body: body); Console.WriteLine($"[*] Request fib({number})"); // //創建消費者用於消息回調 var callbackConsumer = new EventingBasicConsumer(channel); channel.BasicConsume(queue: replyQueue, autoAck: true, consumer: callbackConsumer); callbackConsumer.Received += (model, ea) => { if (ea.BasicProperties.CorrelationId == correlationId) { var responseMsg = $"Get Response: {Encoding.UTF8.GetString(ea.Body)}"; Console.WriteLine($"[x]: {responseMsg}"); } }; Console.ReadLine(); } }

服務端代碼塊

static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "localhost" }; using (var conection = factory.CreateConnection()) { using (var channel = conection.CreateModel()) { channel.QueueDeclare(queue: "rpc_queue", durable: false, exclusive: false, autoDelete: false, arguments: null); var consumer = new EventingBasicConsumer(channel); Console.WriteLine("[*] Waiting for message."); consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); int n = int.Parse(message); Console.WriteLine($"Receive request of Fib({n})"); int result = Fib(n); var properties = ea.BasicProperties; var replyProerties = channel.CreateBasicProperties(); replyProerties.CorrelationId = properties.CorrelationId; channel.BasicPublish(exchange: "", routingKey: properties.ReplyTo, basicProperties: replyProerties, body: Encoding.UTF8.GetBytes(result.ToString())); channel.BasicAck(ea.DeliveryTag, false); Console.WriteLine($"Return result: Fib({n})= {result}"); }; channel.BasicConsume(queue: "rpc_queue", autoAck: false, consumer: consumer); Console.ReadLine(); } } } private static int Fib(int n) { if (n == 0 || n == 1) { return n; } return Fib(n - 1) + Fib(n - 2); }