介紹
Producer：消息的生產者（發送消息的程序）。Queue：消息隊列，理解為一個容器，生產者向它發送消息，它把消息存儲，等待消費者消費。Consumer：消息的消費者（接收消息的程序）。- 此處我們假設 Consumer1、Consumer2、Consumer3 分別為完成任務速度不一樣快的消費者，這會引出此模式的一個重點問題。
如何理解：
工作模式由圖可以看出，就是在簡單隊列模式的基礎上，增加了多個消費者，也就是讓多個消費者綁定同一個隊列，共同去消費，這樣能解決簡單隊列模式中，如果生產速速遠大於消費速度，而導致的消息堆積現象。
- 因為消息被消費后就會消失，所以不必擔心任務會重復執行。
代碼實現：
注：工作隊列模式有兩種
- 輪詢模式：每個消費者均分消息
- 公平分發模式（能者多勞）：按能力分發，處理速度快的分發的多，處理速度慢的分發的少
Worker模式
Worker模式其實就是一對多模式，我們定義兩個消費者來看看效果：
默認情況下，RabbitMQ會順序的將message發給下一個消費者。每個消費者會得到平均數量的message。這種方式稱之為round-robin（輪詢)。
但是很多情況下並不希望消息平均分配,而是要消費快的多消費,消費少的少消費.還有很多情況下一旦其中一個宕機,那么另外接收者的無法接收原本這個接收者所要接收的數據。
我們修改其中一個消費者代碼,讓其等待5秒。在等待中停止運行 看看效果。
consumer.Received += (ch, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($"接受到消息：{message}");
Thread.Sleep(5000);
};
當消費者宕機后消費者1並沒有接受宕機后的數據。所以我們需要消息確認來解決這個問題。
RabbitMQ消息確認
Rabbit中存在兩種消息確認模式
- 自動模式 - 只要消息從隊列獲取,無論消費者獲取到消息后是否成功消費,都認為是消息成功消費。
- 手動模式 - 消費從隊列中獲取消息后,服務器會將該消息處於不可用狀態,等待消費者反饋。如果消費者在消費過程中出現異常，斷開連接且沒有發送應答，那么RabbitMQ會將這個消息重新投遞。
也就是說我們只要將消息確認模式改為手動即可,改為手動確認方式只需改兩處,1.開啟監聽時將autoAck參數改為false,2.消息消費成功后返回確認
修改兩個消費者代碼,並在其中一個中延遲確認。
consumer.Received += (model, ea) =>
{
Thread.Sleep(1000);//等待1秒,
byte[] message = ea.Body;//接收到的消息
Console.WriteLine("接收到信息為:" + Encoding.UTF8.GetString(message));
//返回消息確認
channel.BasicAck(ea.DeliveryTag, true);
};
//消費者開啟監聽
//將autoAck設置false 關閉自動確認
channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
如果在延遲中消費者斷開連接,那么RabbitMQ會重新投遞未確認的消息
能者多勞模式
能者多勞是給消費速度快的消費更多的消息.少的責消費少的消息.能者多勞是建立在手動確認基礎上實現。
在延遲確認的消費中【消費者】添加BasicQos
// 3、創建通道 IModel channel = connection.CreateModel(); // 3.1、每次只能向消費者發送一條信息,再消費者未確認之前,不再向他發送信息 channel.BasicQos(0, 1, false);
可以看出消費快的消費者接受了更多的消息,這就是能者多勞模式的體現
完整測試代碼：
- 生產者實現
using RabbitMQ.Client;
using System;
using System.Text;
namespace RabbitMQTest.Producer
{
/// <summary>
/// RabbitMQ測試_Producer生產者
/// </summary>
class Program
{
static void Main(string[] args)
{
// 1、創建連接工廠
IConnectionFactory factory = new ConnectionFactory()
{
UserName = "admin",
Password = "admin",
HostName = "192.168.1.101", //rabbitmq server ip
Port = 5672,
VirtualHost = "testhost" //在此連接期間要訪問的虛擬主機。 默認值[ / ]
};
// 2、創建連接
IConnection connection = factory.CreateConnection();
// 3、創建通道
IModel channel = connection.CreateModel();
string name = "testQueue";
// 4、聲明一個隊列
channel.QueueDeclare(
queue: name, //消息隊列名稱
durable: false, //是否持久化，true持久化，隊列會保存磁盤,服務器重啟時可以保證不丟失相關信息。
exclusive: false, //是否排他，true排他的，如果一個隊列聲明為排他隊列,該隊列僅對首次聲明它的連接可見,並在連接斷開時自動刪除。
autoDelete: false, //是否自動刪除，true是自動刪除，自動刪除的前提是：致少有一個消費者連接到這個隊列，之后所有與這個隊列連接的消費者都斷開時，才會自動刪除。
arguments: null); //設置隊列的一些其它參數。
Console.WriteLine("\n RabbitMQ連接成功，請輸入消息，輸入exit退出！");
string input;
do
{
input = Console.ReadLine();
byte[] sendBytes = Encoding.UTF8.GetBytes(input);
//發布消息
channel.BasicPublish("", name, null, sendBytes);
//Console.WriteLine("消息發布完畢");
} while (input.Trim().ToLower() != "exit");
Console.WriteLine("\n RabbitMQ測試完畢！");
// 6、關閉通道
channel.Close();
// 7、關閉連接
connection.Close();
}
}
}
- 消費者實現1
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
using System.Threading;
namespace RabbitMQTest.Consumer
{
/// <summary>
/// RabbitMQ測試_Consumer消費者
/// </summary>
class Program
{
static void Main(string[] args)
{
// 1、創建連接工廠
ConnectionFactory factory = new ConnectionFactory()
{
UserName = "admin", //用戶名
Password = "admin", //密碼
HostName = "192.168.1.101", //rabbitmq server ip
Port = 5672, //端口號
VirtualHost = "testhost" //在此連接期間要訪問的虛擬主機。
};
// 2、創建連接
IConnection connection = factory.CreateConnection();
// 3、創建通道
IModel channel = connection.CreateModel();
// 3.1、每次只能向消費者發送一條信息,再消費者未確認之前,不再向他發送信息
channel.BasicQos(0, 1, false);
// 4、事件基本消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
string name = "testQueue";
// 5、接收到消息事件
consumer.Received += (ch, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($"接受到消息：{message}");
Thread.Sleep(5000);
// 6、確認該消費已被消費
channel.BasicAck(ea.DeliveryTag, true);
};
// 7、啟動消費者 設置為自動應答消息
channel.BasicConsume(
queue: name, //消息隊列名稱
autoAck: false, //兩種消息確認模式false 手動模式 true自動模式
consumer: consumer);
Console.WriteLine("消費者1已啟動");
Console.ReadKey();
channel.Close();
connection.Close();
}
}
}
- 消費者實現2
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
namespace RabbitMQTest.Consumer2
{
class Program
{
static void Main(string[] args)
{
// 1、創建連接工廠
ConnectionFactory factory = new ConnectionFactory()
{
UserName = "admin", //用戶名
Password = "admin", //密碼
HostName = "192.168.1.101", //rabbitmq server ip
Port = 5672, //端口號
VirtualHost = "testhost" //在此連接期間要訪問的虛擬主機。
};
// 2、創建連接
IConnection connection = factory.CreateConnection();
// 3、創建通道
IModel channel = connection.CreateModel();
// 4、事件基本消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
string name = "testQueue";
// 5、接收到消息事件
consumer.Received += (ch, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($"接受到消息：{message}");
// 6、確認該消費已被消費
channel.BasicAck(ea.DeliveryTag, true);
};
// 7、啟動消費者 設置為自動應答消息
channel.BasicConsume(name, false, consumer);
Console.WriteLine("消費者2已啟動");
Console.ReadKey();
channel.Close();
connection.Close();
}
}
}