【RabbitMQ】5、RabbitMQ任務分發機制
當有Consumer需要大量的運算時,RabbitMQ Server需要一定的分發機制來balance每個Consumer的load。接下來我們分布講解。
應用場景就是RabbitMQ Server會將queue的Message分發給不同的Consumer以處理計算密集型的任務:
1. Message acknowledgment 消息確認
每個Consumer可能需要一段時間才能處理完收到的數據。如果在這個過程中,Consumer出錯了,異常退出了,而數據還沒有處理完成,那麽 非常不幸,這段數據就丟失了。因為我們采用no-ack的方式進行確認,也就是說,每次Consumer接到數據後,而不管是否處理完 成,RabbitMQ Server會立即把這個Message標記為完成,然後從queue中刪除了。
如果一個Consumer異常退出了,它處理的數據能夠被另外的Consumer處理,這樣數據在這種情況下就不會丟失了(註意是這種情況下)。
為了保證數據不被丟失,RabbitMQ支持消息確認機制,即acknowledgments。為了保證數據能被正確處理而不僅僅是被Consumer收到,那麽我們不能采用no-ack。而應該是在處理完數據後發送ack。
在處理數據後發送的ack,就是告訴RabbitMQ數據已經被接收,處理完成,RabbitMQ可以去安全的刪除它了。
如果Consumer退出了但是沒有發送ack,那麽RabbitMQ就會把這個Message發送到下一個Consumer。這樣就保證了在Consumer異常退出的情況下數據也不會丟失。
這裏並沒有用到超時機制。RabbitMQ僅僅通過Consumer的連接中斷來確認該Message並沒有被正確處理。也就是說,RabbitMQ給了Consumer足夠長的時間來做數據處理。
這樣即使你通過Ctr-C中斷了Recieve.cs,那麽Message也不會丟失了,它會被分發到下一個Consumer。
如果忘記了ack,那麽後果很嚴重。當Consumer退出時,Message會重新分發。然後RabbitMQ會占用越來越多的內存,由於 RabbitMQ會長時間運行,因此這個“內存泄漏”是致命的。去調試這種錯誤,可以通過一下命令打印un-acked Messages.
2. Round-robin dispatching 循環分發
RabbitMQ的分發機制非常適合擴展,而且它是專門為並發程序設計的。如果現在load加重,那麽只需要創建更多的Consumer來進行任務處理即 可。當然了,對於負載還要加大怎麽辦?我沒有遇到過這種情況,那就可以創建多個virtual Host,細化不同的通信類別了。
1、首先開啟兩個Consumer,即運行兩個Recieve.cs。
2、在開啟兩個Producer,即運行兩個Producer.cs。
默認情況下,RabbitMQ 會順序的分發每個Message。當每個收到ack後,會將該Message刪除,然後將下一個Message分發到下一個Consumer。這種分發方式叫做round-robin(優雅分發)。
Producer.cs
class Program { static void Main(string[] args) { ConnectionFactory factory = new ConnectionFactory() { HostName = "localhost" }; using (IConnection connection = factory.CreateConnection()) { using (IModel channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null); var message = GetMessage(args); var body = Encoding.UTF8.GetBytes(message); var properties = channel.CreateBasicProperties(); properties.DeliveryMode = 2;//non-persistent (1) or persistent (2) //channel.TxSelect(); channel.BasicPublish("", "hello", properties, body); //channel.TxCommit(); } } } private static string GetMessage(string[] args) { return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!"); } }
Consumer.cs
//#define demo1 #define demo2 using RabbitMQ.Client; using RabbitMQ.Client.Events; using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; namespace ReceiveDemo2 { /// <summary> /// 一個Send和多個Receive的例子, /// 還加上了ack的例子. /// 優雅分發 /// </summary> class Program { static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "localhost" }; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { channel.QueueDeclare("hello", false, false, false, null); var consumer = new QueueingBasicConsumer(channel); #if demo1 channel.BasicConsume("hello", true, consumer);//自動刪除消息 #else channel.BasicConsume("hello", false, consumer);//需要接受方發送ack回執,刪除消息 #endif Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C"); while (true) { var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//掛起的操作 #if demo2 channel.BasicAck(ea.DeliveryTag, false);//與channel.BasicConsume("hello", false, null, consumer);對應 #endif 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"); #if demo2 //channel.BasicAck(ea.DeliveryTag, false);//與channel.BasicConsume("hello", false, null, consumer);對應,這句話寫道40行和49行運行結果就會不一樣.寫到這裏會發生如果輸出[x] Received {0}之後,沒有輸出 [x] Done之前,CTRL+C結束程序,那麽message會自動發給另外一個客戶端,當另外一個客戶端收到message後,執行完49行的命令之後,服務器會刪掉這個message #endif } } } } } }
3. Message durability消息持久化
在上一節中我們知道了即使Consumer異常退出,Message也不會丟失。但是如果RabbitMQ Server退出呢?軟件都有bug,即使RabbitMQ Server是完美毫無bug的(當然這是不可能的,是軟件就有bug,沒有bug的那不叫軟件),它還是有可能退出的:被其它軟件影響,或者系統重啟 了,系統panic了。。。
為了保證在RabbitMQ退出或者crash了數據仍沒有丟失,需要將queue和Message都要持久化。queue的持久化需要在聲明時指定durable=True,修改Producer和Consumer的channel.QueueDeclare代碼,再次強調,Producer和Consumer都應該去創建這個queue,盡管只有一個地方的創建是真正起作用的:
bool durable = true; channel.QueueDeclare("hello", durable, false, false, null);
上述語句執行不會有什麽錯誤,但是確得不到我們想要的結果,原因就是RabbitMQ Server已經維護了一個叫hello的queue,那麽上述執行不會有任何的作用,也就是hello的任何屬性都不會被影響。這一點在上篇文章也討論過。
那麽workaround也很簡單,聲明一個另外的名字的queue,比如名字定位task_hello,或者通過監控http://localhost:15672/,刪除名為“hello”的Queue。
接下來,還需要持久化Message,即在Producer.cs裏面Publish的時候指定一個properties,方式如下:
static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "localhost" }; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { bool durable = true; channel.QueueDeclare("task_queue", durable, false, false, null);//queue的持久化需要在聲明時指定durable=True var message = GetMessage(args); var body = Encoding.UTF8.GetBytes(message); var properties = channel.CreateBasicProperties(); properties.SetPersistent(true);//需要持久化Message,即在Publish的時候指定一個properties, channel.BasicPublish("", "task_hello", properties, body); } } }
關於持久化的進一步討論:
為了數據不丟失,我們采用了:
- 在數據處理結束後發送ack,這樣RabbitMQ Server會認為Message Deliver 成功。
- 持久化queue,可以防止RabbitMQ Server 重啟或者crash引起的數據丟失。
- 持久化Message,理由同上。
但是這樣能保證數據100%不丟失嗎?
答案是否定的。問題就在與RabbitMQ需要時間去把這些信息存到磁盤上,這個time window雖然短,但是它的確還是有。在這個時間窗口內如果數據沒有保存,數據還會丟失。還有另一個原因就是RabbitMQ並不是為每個Message都做fsync:它可能僅僅是把它保存到Cache裏,還沒來得及保存到物理磁盤上。
因此這個持久化還是有問題。但是對於大多數應用來說,這已經足夠了。當然為了保持一致性,你可以把每次的publish放到一個transaction中。這個transaction的實現需要user defined codes。
那麽商業系統會做什麽呢?一種可能的方案是在系統panic時或者異常重啟時或者斷電時,應該給各個應用留出時間去flash cache,保證每個應用都能exit gracefully。
4. Fair dispatch 公平分發
那麽如果有個Consumer工作比較重,那麽就會導致有的Consumer基本沒事可做,有的Consumer卻是毫無休息的機會。那麽,RabbitMQ是如何處理這種問題呢?
通過 BasicQos 方法設置prefetchCount = 1。這樣RabbitMQ就會使得每個Consumer在同一個時間點最多處理一個Message。換句話說,在接收到該Consumer的ack前,他它不會將新的Message分發給它。 設置方法如下:
channel.BasicQos(0, 1, false);
註意,這種方法可能會導致queue滿。當然,這種情況下你可能需要添加更多的Consumer,或者創建更多的virtualHost來細化你的設計。
Consumer.cs
static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "localhost" }; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { bool durable = true; channel.QueueDeclare("task_queue", durable, false, false, null); channel.BasicQos(0, 1, false);//這樣RabbitMQ就會使得每個Consumer在同一個時間點最多處理一個Message。換句話說,在接收到該Consumer的ack前,他它不會將新的Message分發給它。 var consumer = new QueueingBasicConsumer(channel); channel.BasicConsume("task_hello", false, null, consumer);//需要接受方發送ack回執,刪除消息 Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C"); while (true) { var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//掛起的操作 channel.BasicAck(ea.DeliveryTag, false);//與channel.BasicConsume("task_queue", false, null, consumer);對應 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"); } } } }
出處:https://www.cnblogs.com/qiyebao/p/4205626.html
【RabbitMQ】5、RabbitMQ任務分發機制