【c#】RabbitMQ學習文檔（二）Work Queues（工作隊列）
今天開始RabbitMQ教程的第二講，廢話不多說，直接進入話題。 (使用.NET 客戶端 進行事例演示)
在第一個教程中，我們編寫了一個從命名隊列中發送和接收消息的程序。在本教程中，我們將創建一個工作隊列，這個隊列將用於在多個工人之間分配耗時的任務。
工作隊列【又名：任務隊列】背後主要的思想是避免立刻執行耗時的工作任務，並且一直要等到它結束為止。相反，我們規劃任務並晚些執行。我們封裝一個任務作為消息發送到一個命名的消息隊列中，後臺運行的工作線程將獲取任務並且最終執行該任務。當你運行很多的任務的時候他們會 共享工作線程和隊列。
這個概念在Web應用程序中是尤其有用的，異步執行可以在短時間內處理一個復雜Http請求。
1、準備工作
在本系列教程的前一個教程中，我們發送了一個包含“Hello World！”的消息，現在我們發送一個代表復雜任務的字符串。我們不會創建一個真實的任務，比如對圖像文件進行處理或PDF文件的渲染，因此讓我們假裝我們很忙-通過采用Thread.Sleep()功能來實現復雜和繁忙。我們將根據字符串中的點的數量作為它的復雜性，每一個點將占一秒鐘的“工作”。例如，一個假的任務描述Hello…，有三個點，我們就需要三秒。
我們將稍微修改一下我們以前的例子中Send 程序的代碼，允許從命令行發送任意消息。這個程序將把任務發送到我們的消息隊列中，所以我們叫它NewTask：
像教程一，我們需要生成兩個項目。
dotnet new console --name NewTask mv NewTask/Program.cs NewTask/NewTask.cs dotnet new console --name Worker mv Worker/Program.cs Worker/Worker.cs cd NewTask dotnet add package RabbitMQ.Client dotnet restore cd ../Worker dotnet add package RabbitMQ.Client dotnet restorevar message = GetMessage(args); var body = Encoding.UTF8.GetBytes(message); var properties = channel.CreateBasicProperties(); properties.Persistent = true; channel.BasicPublish(exchange: "", routingKey: "task_queue", basicProperties: properties, body: body);
信息數據我們可以從命令行的參數獲得：
private static string GetMessage(string[] args) { return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!"); }
我們的舊Receive.cs代碼也需要一些修改：需要為消息體中每個點都需要消耗一秒鐘的工作，先要計算出消息體內有幾個點號，然後在乘以1000，就是這個復雜消息所消耗的時間，同時表示這是一個復雜任務。RabbitMQ將處理和發送理消息，並且執行這個任務，讓我們拷貝以下代碼黏貼到Worker的項目中，並進行相應的修改：
var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine(" [x] Received {0}", message); int dots = message.Split(‘.‘).Length - 1; Thread.Sleep(dots * 1000); Console.WriteLine(" [x] Done"); }; channel.BasicConsume(queue: "task_queue", noAck: true, consumer: consumer);
我們自己假設的任務的模擬執行時間就是：
int dots = message.Split(‘.‘).Length - 1; Thread.Sleep(dots * 1000);
2、輪詢調度
我們使用任務隊列的好處之一就是使任務可以並行化，增加系統的並行處理能力。如果我們正在建立一個積壓的工作，我們可以緊緊增加更多的Worker實例就可以完成大量工作的處理，修改和維護就很容易。
首先，讓我們同時運行兩個Worker實例。他們都會從隊列中得到消息，但具體如何？讓我想想。
你需要打開三個控制臺的應用程序。兩個控制臺程序將運行Wroker程序。這些控制臺程序將是我們的兩個消費者C1和C2。
# shell 1 cd Worker dotnet run # => [*] Waiting for messages. To exit press CTRL+C # shell 2 cd Worker dotnet run # => [*] Waiting for messages. To exit press CTRL+C
在第三個控制臺應用程序中我們將發布新的任務。只要你已經啟動了消費者程序，你可以看到一些發布的信息：
# shell 3 cd NewTask dotnet run "First message." dotnet run "Second message.." dotnet run "Third message..." dotnet run "Fourth message...." dotnet run "Fifth message....."
讓我們看看交付了什麽東西在Workers:
# shell 1 # => [*] Waiting for messages. To exit press CTRL+C # => [x] Received ‘First message.‘ # => [x] Received ‘Third message...‘ # => [x] Received ‘Fifth message.....‘ # shell 2 # => [*] Waiting for messages. To exit press CTRL+C # => [x] Received ‘Second message..‘ # => [x] Received ‘Fourth message....‘
默認情況下，RabbitMQ將會發送每一條消息給序列中每一個消費者。每個消費者都會得到相同數量的信息。這種分發消息的方式叫做輪詢。我們嘗試這三個或更多的Workers。
3、消息確認
處理一個任務可能需要幾秒鐘。如果有一個消費者開始了一個長期的任務，並且只做了一部分就發生了異常，你可能想知道到底發生了什麽。我們目前的代碼，一旦RabbitMQ發送一個消息給客戶立即從內存中移除。在這種情況下，如果你關掉了一個Worker，我們將失去它正在處理的信息。我們也將丟失發送給該特定員工但尚未處理的所有信息。
但我們不想失去任何任務。如果一個Worker出現了問題，我們希望把這個任務交給另一個Woker。
為了確保消息不會丟失，RabbitMQ支持消息確認機制。ACK（nowledgement）確認消息是從【消息使用者】發送回來告訴RabbitMQ結果的一種特殊消息，確認消息告訴RabbitMQ指定的接受者已經收到、處理，並且RabbitMQ你可以自由刪除它。
如果一個【消費者Consumer】死亡（其通道關閉，連接被關閉，或TCP連接丟失）不會發送ACK，RabbitMQ將會知道這個消息並沒有完全處理，將它重新排隊。如果有其他用戶同時在線，它就會快速地傳遞到另一個【消費者】。這樣你就可以肯定，沒有消息丟失，即使【Worker】偶爾死了或者出現問題。
在沒有任何消息超時；當【消費者】死亡的時候RabbitMQ會重新發送消息。只要是正常的，即使處理消息需要很長很長的時間也會重發消息給【消費者】。
消息確認的機制默認是打開的。在以前的例子中，我們明確地把它們關閉設置noAck（“沒有手動確認”）參數為true。是時候刪除這個標誌了，並且從Worker發送一個適當確認消息，一旦我們完成了工作任務。
var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine(" [x] Received {0}", message); int dots = message.Split(‘.‘).Length - 1; Thread.Sleep(dots * 1000); Console.WriteLine(" [x] Done"); channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); }; channel.BasicConsume(queue: "task_queue", noAck: false, consumer: consumer);
使用這個代碼，我們可以肯定的是，即使你使用Ctrl + C關掉一個正在處理消息的Worker，也不會丟失任何東西。【Worker】被殺死後，未被確認的消息很快就會被退回。
4、忘記確認
忘記調用BasicAck這是一個常見的錯誤。雖然這是一個簡單的錯誤，但後果是嚴重的。消息會被退回時，你的客戶退出（這可能看起來像是隨機的）但是RabbitMQ將會使用更多的內存保存這些任何延遲確認消息。
為了調試這種錯誤，你可以使用rabbitmqctl打印messages_unacknowledged字段值：sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged
如果是在Window環境下，刪除掉sudo字符就可以：rabbitmqctl.bat list_queues name messages_ready messages_unacknowledged
5、持久性的消息
我們已經學會了如何確保即使【消費者】死亡，任務也不會丟失。但是如果RabbitMQ服務器停止了，我們的任務仍然會丟失的。
當RabbitMQ退出或死機會清空隊列和消息，除非你告訴它即使宕機也不能丟失任何東西。要確保消息不會丟失，有兩件事情我們是必需要做的：我們需要將隊列和消息都標記為持久的。
首先，我們需要確保我們RabbitMQ從來都不會損失我們的的隊列。為了做到這一點，我們需要聲明我們的隊列為持久化的：
channel.QueueDeclare(queue: "hello", durable: true, exclusive: false, autoDelete: false, arguments: null);
雖然這個命令本身是正確的，它不會起作用在我們目前的設置中。這是因為我們已經定義了一個叫hello的隊列，它不是持久化的。RabbitMQ不允許你使用不同的參數重新定義一個已經存在的隊列，在任何程序代碼中，都試圖返回一個錯誤。但有一個快速的解決方法-讓我們聲明一個名稱不同的隊列，例如task_queue：
channel.QueueDeclare(queue: "task_queue", durable: true, exclusive: false, autoDelete: false, arguments: null);
這行代碼QueueDeclare表示隊列的聲明，創建並打開隊列，這個段代碼需要應用到【生產者】和【消費者】中。
在這一點上，我們相信，task_queue隊列不會丟失任何東西即使RabbitMQ重啟了。現在我們要通過設置IbasicProperties.SetPersistent屬性值為true來標記我們的消息持久化的。
var properties = channel.CreateBasicProperties();
properties.Persistent = true;
關於消息持久性的註意
將消息標記為持久性並不能完全保證消息不會丟失。雖然該設置告訴RabbitMQ時時刻刻把保存消息到磁盤上，但是這個時間間隔還是有的，當RabbitMQ已經接受信息但並沒有保存它，此時還有可能丟失。另外，RabbitMQ不會為每個消息調用fsync（2）--它可能只是保存到緩存並沒有真正寫入到磁盤。雖然他的持久性保證不強，但它我們簡單的任務隊列已經足夠用了。如果您需要更強的保證，那麽您可以使用Publisher Comfirms。
6、公平調度
你可能已經註意到，調度仍然沒有像我們期望的那樣的工作。例如，在兩個Workers的情況下，當所有的奇數消息是沈重的，甚至消息是輕的，一個Worker忙個不停，而另一個Worker幾乎沒事可做。哎，RabbitMQ對上述情況一無所知，仍將消息均勻發送。
發生這種情況是因為當有消息進入隊列的時候RabbitMQ才僅僅調度了消息。它根本不看【消費者】未確認消息的數量，它只是盲目的把第N個消息發送給第N個【消費者】。
為了避免上述情況的發生，我們可以使用prefetchcount = 1的設置來調用BasicQos方法。這個方法告訴RabbitMQ在同一時間不要發送多余一個消息的數據給某個【Worker】。或者，換句話說，當某個消息處理完畢，並且已經收到了消息確認之後，才可以繼續發送消息給那個【Worker】。相反，它將把消息分配給給下一個不忙的【Worker】。
channel.BasicQos(0, 1, false);
註意隊列大小
如果所有的工人都很忙，你的隊列可以填滿。你要留意這一點，也許會增加更多的【Worker】，或者有其他的策略。
7、把所有的代碼放在一起
NewTask.cs類最終的代碼是：
1 using System; 2 using RabbitMQ.Client; 3 using System.Text; 4 5 class NewTask 6 { 7 public static void Main(string[] args) 8 { 9 var factory = new ConnectionFactory() { HostName = "localhost" }; 10 using(var connection = factory.CreateConnection()) 11 using(var channel = connection.CreateModel()) 12 { 13 channel.QueueDeclare(queue: "task_queue", 14 durable: true, 15 exclusive: false, 16 autoDelete: false, 17 arguments: null); 18 19 var message = GetMessage(args); 20 var body = Encoding.UTF8.GetBytes(message); 21 22 var properties = channel.CreateBasicProperties(); 23 properties.Persistent = true; 24 25 channel.BasicPublish(exchange: "", 26 routingKey: "task_queue", 27 basicProperties: properties, 28 body: body); 29 Console.WriteLine(" [x] Sent {0}", message); 30 } 31 32 Console.WriteLine(" Press [enter] to exit."); 33 Console.ReadLine(); 34 } 35 36 private static string GetMessage(string[] args) 37 { 38 return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!"); 39 } 40 }
Worker.cs完整源碼如下：
1 using System; 2 using RabbitMQ.Client; 3 using RabbitMQ.Client.Events; 4 using System.Text; 5 using System.Threading; 6 7 class Worker 8 { 9 public static void Main() 10 { 11 var factory = new ConnectionFactory() { HostName = "localhost" }; 12 using(var connection = factory.CreateConnection()) 13 using(var channel = connection.CreateModel()) 14 { 15 channel.QueueDeclare(queue: "task_queue", 16 durable: true, 17 exclusive: false, 18 autoDelete: false, 19 arguments: null); 20 21 channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false); 22 23 Console.WriteLine(" [*] Waiting for messages."); 24 25 var consumer = new EventingBasicConsumer(channel); 26 consumer.Received += (model, ea) => 27 { 28 var body = ea.Body; 29 var message = Encoding.UTF8.GetString(body); 30 Console.WriteLine(" [x] Received {0}", message); 31 32 int dots = message.Split(‘.‘).Length - 1; 33 Thread.Sleep(dots * 1000); 34 35 Console.WriteLine(" [x] Done"); 36 37 channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false); 38 }; 39 channel.BasicConsume(queue: "task_queue", 40 noAck: false, 41 consumer: consumer); 42 43 Console.WriteLine(" Press [enter] to exit."); 44 Console.ReadLine(); 45 } 46 } 47 }
使用消息確認和BasicQos方法可以建立一個工作隊列。持久化的選項可以讓我們的任務隊列保持存活即使RabbitMQ重啟。
好了，寫完了，翻譯的不好，大家見諒。
原文地址如下：http://www.rabbitmq.com/tutorials/tutorial-two-dotnet.html
歡迎大家來探討。
【c#】RabbitMQ學習文檔（二）Work Queues（工作隊列）