1 消息隊列的優點
消息隊列本質是生產者——消費者模式。也有很多使用方式。那么它有什么優點呢？
以日常生活中郵寄信件這個行為舉例，
當只有1個寄信人，1個郵遞員的時候。寄信人想要寄信，到指定地點（郵局），直接將信件交給郵遞員即可。
當有50個寄信人，1個郵遞員的時候。這50個寄信人就要依次排隊等待郵遞員處理信件。
可以增加郵遞員的數量，但是依然會有忙閑不均的問題存在。
我們現在增加一個郵筒（也就是數據緩沖區）
在這個例子中，寄信人就是生產者，郵遞員是消費者。而郵筒就是一個消息隊列。這個郵筒解決了以下問題：
1.1 解除耦合
實現了時間上解耦，也實現了對象間解耦。
之前郵遞員隸屬於A郵局，寄信人想要寄信，到指定地點，直接將信件交給郵遞員即可。如果因為實際需求，以后由B郵局的快遞員負責寄信業務。那么寄信人就要去另一個地點寄信。
這就是由於耦合產生的問題。
現在不管信件是由A郵局還是其他郵局負責，寄信人只管將信件投遞進郵筒就行了。解除了寄信人和郵遞員的耦合性。
1.2 實現異步處理
之前寄信將信件直接交給郵遞員，可能要等待郵遞員要確認很多信息（比如寄件人信息）之后，長輒幾分鍾，才能結束本次寄信的行為。
而現在將信件直接投遞到郵箱里，只要不到1S，就能結束寄信的行為。
1.3 支持並發操作
解決同步處理的阻塞問題。
之前所有寄信人需要排隊等待上一個人寄信完畢，才能開始寄信。
現在所有寄信人都把信件投遞進郵筒即可。
1.4 實現流量削峰
可以根據郵遞員方的處理能力，調節郵筒的容量。超過這個容量后，郵筒就放不下（拒絕）信件了。
即能根據下游的處理能力自由調節流量，實現削峰。
2 安裝erlang和RabbitMQ
2.1 安裝erlang
由於RabbitMQ是基於erlang開發的，需要先安裝erlang。
確認自己要安裝的RabbitMQ依賴的erlang的最低版本。
erlang:https://www.erlang.org/downloads
安裝后添加環境變量。
在系統變量中添加：
變量名：ERLANG_HOME
變量值：C:\Program Files\erl-24.0（安裝ERLANG的文件夾）
然后在用戶變量的PATH中添加：%ERLANG_HOME%\bin
添加完環境變量之后可能需要重啟。
然后打開CMD，運行erl，出現版本號為成功。
2.2 安裝RabbitMQ
RabbitMQ:https://github.com/rabbitmq/rabbitmq-server/releases/
安裝成功后會自動創建RabbitMQ服務並且啟動
可以在任務管理器中確認：
2.3 安裝RabbitMQ的Web管理插件
在命令行中CD到安裝目錄下，執行
rabbitmq-plugins.bat enable rabbitmq_management
成功后進入瀏覽器，輸入：http://localhost:15672
初始賬號和密碼：guest/guest
3 理解消息隊列中的基本概念
消息隊列中有Exchange、Connection、Channel、Queue等概念
3.1 Exchange（交換機）
是生產者和消息隊列的一個中介，負責將生產者的消息分發給消息隊列。如果使用簡單模式（只有一個生產者，一個消費者，一對一）時，不配置Exchange，實際上使用的是默認的Exchange。
3.2 Connection（連接）
是連接到MQ的TCP連接。為了方便理解，可以將Connection想象成一個光纖電纜。
3.3 Channel（通道）
一個Connection中存在多個Channel。可以把Channel理解為光纖電纜中的光纖。
3.4 Queue（消息隊列）
一個Channel中可以存在多個Queue。
3.5 Broker（代理交換節點）
一個Broker就是一個RabbitMQ服務節點，包含多個Exchange（交換機）和Queue（消息隊列）。
3.6 其他
因為建立和銷毀 TCP 連接是非常昂貴的開銷，所以一般維持Connection。在Connection之上，操作channel。
Channel的其中一個作用就是，屏蔽Connection的TCP層面的細節，方便開發，同時達到TCP連接復用的效果。
4 嘗試消息隊列的簡單模式（一對一）
特點：一個生產者對應一個消費者。最簡單的模式。
場景：一對一私聊。
新建一個解決方案，包含兩個控制台程序，分別是生產者和消費者。
右鍵解決方案，設置多項目啟動。
4.1 生產者代碼
/// <summary>
/// 生產者
/// </summary>
internal class Program
{
private static void Main(string[] args)
{
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
//創建RabbitMQ的TCP長連接（可以比喻成一個光纖電纜）
//因為建立和銷毀 TCP 連接是非常昂貴的開銷，所以一般維持連接（TCP連接復用）。在連接之上，建立和銷毀channel。
var connection = factory.CreateConnection();
//創建通道（可以比喻成光纖電纜中的"一根"光纖）
var channel = connection.CreateModel();
/*聲明一個隊列：實現通道與隊列的綁定
* 5個參數：
* queue:被綁定的消息隊列名，當該消息隊列不存在時，將新建該消息隊列
* durable:是否使用持久化
* exclusive：該通道是否獨占該隊列
* autoDelete：消費完成時是否刪除隊列， 該刪除操作在消費者徹底斷開連接之后進行。
* args：其他配置參數
*/
channel.QueueDeclare("hello", false, false, false, null);
Console.WriteLine("\nRabbitMQ連接成功，生產者已啟動，請輸入消息，輸入exit退出！");
string input;
do
{
input = Console.ReadLine();
var sendBytes = Encoding.UTF8.GetBytes(input);
//發布消息
channel.BasicPublish("", "hello", null, sendBytes);
}
while (input.Trim().ToLower() != "exit");
channel.Close();
connection.Close();
}
}
4.2 消費者代碼
/// <summary>
/// 消費者
/// </summary>
internal class Program
{
private static void Main(string[] args)
{
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
//創建連接
var connection = factory.CreateConnection();
//創建通道
var channel = connection.CreateModel();
//事件基本消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//接收到消息事件
consumer.Received += (ch, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($"收到消息： {message}");
//確認該消息已被消費
channel.BasicAck(ea.DeliveryTag, false);
};
//啟動消費者 設置為手動應答消息
channel.BasicConsume("hello", false, consumer);
Console.WriteLine("消費者已啟動");
Console.ReadKey();
channel.Dispose();
connection.Close();
}
}
4.3 測試
兩個項目一起啟動之后
在生產者對應的控制台輸入文字后，添加到消息隊列中，由消費者進行消費，顯示在消費者控制台上。
5 嘗試消息隊列的WORK模式
特點：爭奪消息，能者多勞。每個消費者獲得的消息具有唯一性。
場景：搶紅包。搶單。
5.1 生產者的代碼
為了代碼邏輯清晰，將各種模式的代碼從Main函數中提出來單獨封裝成函數。Main函數中使用Switch來方便之后的測試。
/// <summary>
/// 生產者
/// </summary>
internal static class Program
{
private static void Main(string[] args)
{
//選擇的模式類型
string ModeNumber = "2";
switch (ModeNumber)
{
case "1":
SignalMode();
break;
case "2":
WorkMode();
break;
default:
break;
}
}
/// <summary>
/// 簡單模式
/// </summary>
private static void SignalMode()
{
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
//創建RabbitMQ的TCP長連接（可以比喻成一個光纖電纜）
//因為建立和銷毀 TCP 連接是非常昂貴的開銷，所以一般維持連接（TCP連接復用）。在連接之上，建立和銷毀channel。
var connection = factory.CreateConnection();
//創建通道（可以比喻成光纖電纜中的"一根"光纖）
var channel = connection.CreateModel();
/*聲明一個隊列：實現通道與隊列的綁定
* 5個參數：
* queue:被綁定的消息隊列名，當該消息隊列不存在時，將新建該消息隊列
* durable:是否使用持久化
* exclusive：該通道是否獨占該隊列
* autoDelete：消費完成時是否刪除隊列， 該刪除操作在消費者徹底斷開連接之后進行。
* args：其他配置參數
*/
channel.QueueDeclare("隊列A", false, false, false, null);
Console.WriteLine("\nRabbitMQ連接成功，生產者已啟動，請輸入消息，輸入exit退出！");
string input;
do
{
input = Console.ReadLine();
var sendBytes = Encoding.UTF8.GetBytes(input);
//發布消息
channel.BasicPublish("", "hello", null, sendBytes);
}
while (input.Trim().ToLower() != "exit");
channel.Close();
connection.Close();
}
/// <summary>
/// Work模式
/// </summary>
private static void WorkMode()
{
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
channel.QueueDeclare("隊列A", false, false, false, null);
for (int i = 0; i < 50; i++)
{
String message = "" + i;
var sendBytes = Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "隊列A", null, sendBytes);
Console.WriteLine(" [x] Sent '" + message + "'");
Thread.Sleep(i * 10);
}
channel.Close();
connection.Close();
}
}
5.2 消費者的代碼
為了模擬多個消費者爭奪消息，將之前的消費者項目重命名為"RabbitMQ_Consumer01",並新建項目"RabbitMQ_Consumer02"。在work模式中，消費者01和消費者02的代碼是相同的。
並將生產者、消費者01、消費者02同時設為啟動項（由於消費者代碼相同，只貼一個）。
/// <summary>
/// 消費者01
/// </summary>
internal static class Program
{
private static void Main(string[] args)
{
//選擇的模式類型
string ModeNumber = "2";
switch (ModeNumber)
{
case "1":
SignalMode();
break;
case "2":
WorkMode();
break;
default:
break;
}
}
/// <summary>
/// 簡單模式
/// </summary>
private static void SignalMode()
{
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
//創建連接
var connection = factory.CreateConnection();
//創建通道
var channel = connection.CreateModel();
//事件基本消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//接收到消息事件
consumer.Received += (model, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($@"收到消息： {message}");
//確認該消息已被消費
channel.BasicAck(ea.DeliveryTag, false);
};
//啟動消費者 設置為手動應答消息
channel.BasicConsume("隊列A", false, consumer);
Console.WriteLine($@"消費者已啟動");
Console.ReadKey();
channel.Dispose();
connection.Close();
}
/// <summary>
/// Work模式
/// </summary>
private static void WorkMode()
{
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
channel.QueueDeclare("隊列A", false, false, false, null);
/** 設置限流機制
* param1: prefetchSize，消息本身的大小 如果設置為0 那么表示對消息本身的大小不限制
* param2: prefetchCount，告訴rabbitmq，不要一次性給消費者推送大於N個消息（一旦有N個消息沒有Ack，此消費者不再獲取消息，直到有消息Ack為止）
* param3：global，是否將上面的設置應用於整個通道，false表示只應用於當前消費者
*/
channel.BasicQos(0, 1, false);
// 定義隊列的消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//啟動消費者
//false為手動應答，true為自動應答
channel.BasicConsume("隊列A", false, consumer);
consumer.Received += (model, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($@"收到消息： {message}");
//確認該消息已被消費，手動返回完成
channel.BasicAck(ea.DeliveryTag, false);
};
Console.WriteLine($@"消費者01已啟動");
Console.ReadKey();
channel.Dispose();
connection.Close();
}
}
5.3 測試處理能力相同的情況
啟動項目后，可以看到消費者向隊列推送消息1-50。而消費者01和消費者02對隊列中的消息進行爭搶，並且獲取的消息具有唯一性。
可以看到由於消費者01、02的處理能力相同，爭搶消息的數量也是平均的。
5.4 測試處理能力不相同的情況
那如何體現“能者多勞”這個特點呢。
在消費者01獲取消息后，通過Thread.Sleep(1000);模擬消費者01的對消息的處理速度比較慢
可以看到，由於消費者01的處理速度慢，爭搶到的消息也比較少
6 消費者端的限流配置（QOS）
6.1 配置限流參數
消費者中的這句代碼就是在配置限流參數：
channel.BasicQos(0, 1, false);
param1: prefetchSize，消息本身的大小 如果設置為0 那么表示對消息本身的大小不限制
param2: prefetchCount，告訴rabbitmq，不要一次性給消費者推送大於N個消息（簡單點說就是：一旦有N個消息沒有Ack，此消費者不再獲取消息，直到有消息Ack為止）
param3: global，是否將上面的設置應用於整個通道，false表示只應用於當前消費者
6.2 測試限流功能
我們將消費者02的限流數量設置為1，同時注釋掉手動ACK的語句。
這樣消費者02獲取消息后，由於不會進行ACK操作，會導致消費者02的阻塞。我們設置的限流數量是1，所以消費者02由始至終只會獲得一條消息。
7 RabbitMQ的確認機制（ACK）
ACK是acknowledgment的縮寫。
隊列接收到消費者的ACK信息后，才會將對應的消息進行刪除操作。
RabbitMQ的確認方式分為自動ACK和手動ACK。
7.1 自動ACK和手動ACK的區別
自動ACK：消費者獲取到消息后，會自動進行ACK操作。
手動ACK：可以自定義調用ACK操作的位置。
選擇自動ACK，如果消費者處理時出現問題，或者中途退出沒有處理。但隊列已經接收到自動ACK把消息刪除了，可能導致對消息處理出錯。
選擇手動ACK，可以將ACK的時機放在消費者正確將消息處理完畢之后。如果消費者中途退出，消息會由另一個消費者獲取到進行操作。
7.2 如何配置ACK模式
消費者中這行代碼的第二個參數就是在配置ACK模式(bool AutoAck)
channel.BasicConsume("隊列A", false, consumer);
如果選擇手動ACK，就要選擇時機執行channel.BasicAck()函數。
7.3 測試手動ACK模式
在消費者02,接收到消息的事件委托函數中，增加以下代碼。
當消費者02首次獲取到大於10的數時，模擬消費失敗，消費者02退出的的場景。
可以看到，消費者02獲取到消息"12"之后，在進行ACK操作之前就退出了。消息再次由消費者01獲得。
8 嘗試消息隊列的發布/訂閱模式
特點：每個消費者有各自的隊列，獲取的消息相同。
場景：群聊天。
生產者向交換機發送消息，交換機將消息廣播到各個隊列。
8.1 交換機的類型
交換機有四種類型，分別是fanout、direct、topic、headers。
其中fanout（扇形交換）就是發布/訂閱模式需要用到的。
（其實fanout交換機是Direct交換機的簡化版，對於Direct先不進行討論）
8.2 生產者的代碼
Main函數中增加發布訂閱類型，並選擇
private static void Main(string[] args)
{
//選擇的模式類型
string ModeNumber = "3";
switch (ModeNumber)
{
case "1":
SignalMode();
break;
case "2":
WorkMode();
break;
case "3":
PubSubMode();
break;
default:
break;
}
}
增加發布/訂閱類型的函數
/// <summary>
/// 發布/訂閱模式
/// </summary>
private static void PubSubMode()
{
//聲明接受消息的交換機的名稱
string myExangeName = "ExangeTypeF";
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
/** 聲明交換機
* param1: 接受消息的交換機名稱
* param2: 交換機模式
*/
channel.ExchangeDeclare(myExangeName, "fanout");
Console.WriteLine("生產者已啟動");
for (int i = 0; i < 50; i++)
{
string message = "" + i;
var sendBytes = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(myExangeName, "", null, sendBytes);
Console.WriteLine(" [x] Sent '" + message + "'");
Thread.Sleep(i * 10);
}
channel.Close();
connection.Close();
Console.ReadKey();
}
8.3 消費者01的代碼
Main函數中增加發布訂閱類型，並選擇
private static void Main(string[] args)
{
//選擇的模式類型
string ModeNumber = "3";
switch (ModeNumber)
{
case "1":
SignalMode();
break;
case "2":
WorkMode();
break;
case "3":
PubSubMode();
break;
default:
break;
}
}
增加發布/訂閱類型的函數
/// <summary>
/// 發布/訂閱模式
/// </summary>
private static void PubSubMode()
{
//聲明接受消息的交換機的名稱
string myExangeName = "ExangeTypeF";
//聲明監聽的隊列
string myQueueName = "Queue01";
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
//聲明隊列
channel.QueueDeclare(myQueueName, false, false, false, null);
//綁定隊列到交換機
channel.QueueBind(myQueueName, myExangeName, "");
channel.BasicQos(0, 1, false);
// 定義隊列的消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//啟動消費者
//false為手動應答，true為自動應答
channel.BasicConsume(myQueueName, false, consumer);
consumer.Received += (model, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($@"收到消息： {message}");
//確認該消息已被消費，手動返回完成
channel.BasicAck(ea.DeliveryTag, false);
};
Console.WriteLine($@"消費者01已啟動");
Console.ReadKey();
channel.Dispose();
connection.Close();
}
8.4 消費者02的代碼
和消費者01基本的一致。注意監聽的隊列名稱要和消費者01監聽的隊列區分開。
否則兩個消費者監聽同一個隊列，又變成WORK模式了。
string myQueueName = "Queue02";
8.5 測試
項目啟動后，可以看到，兩個消費者監聽的兩個隊列，都接受到了交換機發送的消息廣播。
9 嘗試消息隊列的路由模式
特點：在發布/訂閱模式的基礎上，增加路由鍵值（routingkey），達到選擇性向隊列發送消息的目的。
9.1 交換機類型
實現路由模式的交換機類型為Direct（直連交換機）。交換機將消息推送到綁定着對應路由鍵值的隊列中。
9.2 生產者的代碼
Main函數中增加路由類型，並選擇
private static void Main(string[] args)
{
//選擇的模式類型
string ModeNumber = "4";
switch (ModeNumber)
{
case "1":
SignalMode();
break;
case "2":
WorkMode();
break;
case "3":
PubSubMode();
break;
case "4":
RoutingMode();
break;
default:
break;
}
}
增加路由模式相關代碼，並設定區分路由鍵值的邏輯。
如果數字大於15且小於30，路由鍵值為X1；否則，路由鍵值為X2。
/// <summary>
/// 路由模式
/// </summary>
private static void RoutingMode()
{
//聲明接受消息的交換機的名稱
string myExangeName = "ExangeTypeR";
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
/** 聲明交換機
* param1: 接受消息的交換機名稱
* param2: 交換機模式
*/
channel.ExchangeDeclare(myExangeName, "direct");
Console.WriteLine("生產者已啟動");
for (int i = 0; i < 50; i++)
{
string message = "" + i;
var sendBytes = Encoding.UTF8.GetBytes(message);
//區分設定路由鍵值
//如果大於15且小於30，路由鍵值為X1；否則，路由鍵值為X2
if (i > 15 && i < 30)
{
/** 發布消息
* param1: 接受消息的交換機名稱
* param2: 路由鍵值
* param3: 其他參數（暫時用不到）
* param4: 二進制的消息體
*/
channel.BasicPublish(myExangeName, "X1", null, sendBytes);
}
else
{
channel.BasicPublish(myExangeName, "X2", null, sendBytes);
}
Console.WriteLine(" [x] Sent '" + message + "'");
Thread.Sleep(i * 10);
}
channel.Close();
connection.Close();
Console.ReadKey();
}
9.3 消費者01的代碼
Main函數中增加路由類型，並選擇（和生產者的Main函數一樣，不粘貼了）
增加路由模式相關代碼：
/// <summary>
/// 路由模式
/// </summary>
private static void RoutingMode()
{
//聲明接受消息的交換機的名稱
string myExangeName = "ExangeTypeR";
//聲明監聽的隊列
string myQueueName = "Queue01";
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
//聲明隊列
channel.QueueDeclare(myQueueName, false, false, false, null);
//綁定隊列到交換機
channel.QueueBind(myQueueName, myExangeName, "X1");
channel.BasicQos(0, 1, false);
// 定義隊列的消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//啟動消費者
//false為手動應答，true為自動應答
channel.BasicConsume(myQueueName, false, consumer);
consumer.Received += (model, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($@"收到消息： {message}");
//確認該消息已被消費，手動返回完成
channel.BasicAck(ea.DeliveryTag, false);
};
Console.WriteLine($@"消費者01已啟動");
Console.ReadKey();
channel.Dispose();
connection.Close();
}
注意這句話最后一個參數，綁定了隊列匹配的路由鍵值
//綁定隊列到交換機
channel.QueueBind(myQueueName, myExangeName, "X1");
9.4 消費者02的代碼
Main函數中增加路由類型，並選擇（和生產者的Main函數一樣，不粘貼了）。
新增路由模式的代碼：
和消費者01的代碼相同，只修改了兩處。
監聽的隊列為“Queue02”：
//聲明監聽的隊列
string myQueueName = "Queue02";
隊列綁定的路由鍵值為“X2”：
//綁定隊列到交換機
channel.QueueBind(myQueueName, myExangeName, "X2");
9.5 測試
可以看到路由鍵值為X1的消息（大於15且小於30），都被直連交換機轉發到匹配着“X1”路由鍵值的“Queue1”隊列中，發送給消費者01。
10 嘗試消息隊列的主題模式
主題模式是路由模式的進化型。
如果說路由模式中交換機發送消息的依據是匹配着路由鍵值的隊列，
那么主題模式中發送消息的依據則是根據通配符找到符合條件的隊列，進行消息發送。
有些類似於抖音或微博中的#，話題功能。
說的簡單一點，
路由模式是“全字匹配”路由鍵值，
主題模式是根據規則“模糊查詢”路由鍵值。
10.1 交換機的類型
實現主題模式的交換機類型為topic（主題交換機）。
10.2 星花*與井號#的效果
主題模式中的路由鍵值是由多個主題組成，由"."進行分割。
例如"it.computer.cpu"。
消費者進行匹配的規則有兩種，星花*與井號#。
星花*的效果是只能忽略一個主題進行匹配。
井號#的效果是可以忽略多個主題進行匹配。
當一個隊列的路由鍵值為"it.*"時，是"接收不到"消息的。
當一個隊列的路由鍵值為"it.computer.*"時，是"可以收到"消息的。
當一個隊列的路由鍵值為"it.#"時，是"可以收到"消息的。
10.3 生產者的代碼
Main函數中增加發布主題類型，並選擇
private static void Main(string[] args)
{
//選擇的模式類型
string ModeNumber = "5";
switch (ModeNumber)
{
case "1":
SignalMode();
break;
case "2":
WorkMode();
break;
case "3":
PubSubMode();
break;
case "4":
RoutingMode();
break;
case "5":
TopicMode();
break;
default:
break;
}
}
添加主題模式的代碼
/// <summary>
/// 主題模式
/// </summary>
private static void TopicMode()
{
//聲明接受消息的交換機的名稱
string myExangeName = "ExangeTypeT";
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
channel.ExchangeDeclare(myExangeName, "topic");
Console.WriteLine("生產者已啟動");
for (int i = 0; i < 50; i++)
{
string message = "" + i;
var sendBytes = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(myExangeName, "it.computer.cpu", null, sendBytes);
Console.WriteLine(" [x] Sent '" + message + "'");
Thread.Sleep(i * 10);
}
channel.Close();
connection.Close();
Console.ReadKey();
}
和路由模式的主要區別在於：
第二個參數路由鍵值，填寫的是多個單詞（話題）拼接而成的字符串，用“.”做分隔。
channel.BasicPublish(myExangeName, "it.computer.cpu", null, sendBytes);
10.4 消費者01的代碼
Main函數中增加主題模式，並選擇（和生產者的Main函數一樣，不粘貼了）。
新增主題模式相關代碼
/// <summary>
/// 主題模式
/// </summary>
private static void TopicMode()
{
//聲明接受消息的交換機的名稱
string myExangeName = "ExangeTypeT";
//聲明監聽的隊列
string myQueueName = "Queue01";
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
//聲明隊列
channel.QueueDeclare(myQueueName, false, false, false, null);
//綁定隊列到交換機
channel.QueueBind(myQueueName, myExangeName, "it.computer.*");
channel.BasicQos(0, 1, false);
// 定義隊列的消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//啟動消費者
//false為手動應答，true為自動應答
channel.BasicConsume(myQueueName, false, consumer);
consumer.Received += (model, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($@"收到消息： {message}");
//確認該消息已被消費，手動返回完成
channel.BasicAck(ea.DeliveryTag, false);
};
Console.WriteLine($@"消費者01已啟動");
Console.ReadKey();
channel.Dispose();
connection.Close();
}
10.5 消費者02的代碼
Main函數中增加主題模式，並選擇（和生產者的Main函數一樣，不粘貼了）。
新增主題模式相關代碼
/// <summary>
/// 主題模式
/// </summary>
private static void TopicMode()
{
//聲明接受消息的交換機的名稱
string myExangeName = "ExangeTypeT";
//聲明監聽的隊列
string myQueueName = "Queue02";
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
//聲明隊列
channel.QueueDeclare(myQueueName, false, false, false, null);
//綁定隊列到交換機
channel.QueueBind(myQueueName, myExangeName, "it.#");
channel.BasicQos(0, 1, false);
// 定義隊列的消費者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//啟動消費者
//false為手動應答，true為自動應答
channel.BasicConsume(myQueueName, false, consumer);
consumer.Received += (model, ea) =>
{
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($@"收到消息： {message}");
//確認該消息已被消費，手動返回完成
channel.BasicAck(ea.DeliveryTag, false);
};
Console.WriteLine($@"消費者02已啟動");
Console.ReadKey();
channel.Dispose();
connection.Close();
}
消費者01和消費者02，監聽的主題是不同的。
消費者01：channel.QueueBind(myQueueName, myExangeName, "it.computer.*");
消費者02：channel.QueueBind(myQueueName, myExangeName, "it.#");
10.6 測試
可以看到消費者01和消費者02都可以接收到消息。
11 實現消息的持久化
11.1 消息持久化的作用
消息持久化的目的是在服務器端保存未消費的消息，防止服務器宕機或者消息隊列服務因故關閉導致的消息丟失，和手動ACK機制一樣，都可以用來提高消息隊列服務的可靠性。
RabbitMQ中為了最終達到持久化的目的，需要將3個部分都設置為持久化。分別是交換機持久化、隊列持久化、消息持久化。
（測試消息持久化，基於生產者和消費者01的主題模式代碼進行修改。不涉及消費者02）
11.2 生產者的代碼
交換機持久化：
聲明交換機時，將第三個可選參數durable設為true（默認為false），實現交換機持久化：
//開啟交換機持久化
channel.ExchangeDeclare(myExangeName, "topic", true);
消息持久化：
調用接口新建基礎參數類，將DeliveryMode設為2。
發送消息時，將基礎參數類作為參數傳入：
IBasicProperties properties = channel.CreateBasicProperties();
properties.DeliveryMode = 2;
channel.BasicPublish(myExangeName, "it.computer.cpu", properties, sendBytes);
11.3 消費者01的代碼
隊列持久化：
聲明隊列時將第二個參數durable設為true，實現隊列持久化：
//聲明隊列
//開啟隊列持久化
channel.QueueDeclare(myQueueName, true, false, false, null);
11.4 重啟服務相關指令
Windos下RabbitMQ關閉和啟動的指令分別為：
rabbitmq-service start 開始服務
rabbitmq-service stop 停止服務
11.5 測試持久化
減慢消費速度
為了隊列中的消息不會太快被消費，我們在消費者01中，將消費的速度減慢。
增加代碼Thread.Sleep(6000);
設置生產者和消費者01同時啟動
開始生產消息
當生產者將所有消息都發送后，消費者01沒有全部消費完之前，關閉程序。
（注：截圖后消費者01消費到消息17了，沒有截取到）
重啟RabbitMQ服務
rabbitmq-service stop 停止服務
rabbitmq-service start 開始服務
測試持久化效果
將解決方案設置為消費者01單項目啟動
啟動消費者01，此時由於消息持久化成功，消費者01會繼續消費關閉服務前未被消費的消息。
12 保證消息的冪等性
12.1 什么是消息的冪等性
如果同一個消息，因為各種原因，不慎被消費了多次（例如多次點按按鈕），和只消費一次得到的數據是相同的。就可以說保持了冪等性。
RabbitMQ是沒有辦法自己解決冪等性問題的，甚至某些情況下會造成消息重復消費的問題。
例如在第7.3節中，我們嘗試了RabbitMQ的ACK機制。
可以看到，消費者02獲取到消息"12"之后，其實已經將消息消費完了（輸出在控制台），只是在進行ACK操作之前模擬了消費者出錯退出。導致消息再次由消費者01獲得並消費。
最終，這條消息就被消費了兩次。
如果我們不人為保證消息的冪等性，數據就會出錯。
關於消息的等冪性，請看另一篇文章：
13 防止消息丟失
當通過RabbitMQ傳送消息時，如何防止消息半路丟失？
消費者導致消息丟失
詳見第七節，我們可以使用消費者的手動ACK功能。當一個消費者接受到消息后沒有進行ACK操作就掉線，交換機會將這條消息發送到另一個消費者監聽的隊列中。
RabbitMQ導致消息丟失
詳見第十一節，我們可以通過啟用RabbitMQ的持久化功能，保證消息服務宕掉重啟后，未消費完成的消息依然存在，並向消費者發送。
生產者導致丟失
生產者在發送消息后，是無法得知消息是否正確發送到隊列服務。
我們可以通過使用RabbitMQ的事務或通道（Channel）的Confirm功能預防這種情況。
Confirm類似於消費者的ACK，是隊列服務提供給生產者的ACK。
官方文檔：https://www.rabbitmq.com/tutorials/tutorial-seven-dotnet.html
13.1 WaitForConfirms()
使用channel.WaitForConfirms()函數來獲取Borker的確認消息。
確認消息有兩種結果，ack'd（接收到）和nack'd（丟失）。官方文檔對nack'd的解釋是:"meaning the broker could not take care of it for some reason"。
如果狀態為ack'd，WaitForConfirms()的返回值為true。
如果狀態為nack'd，或者在規定時間內沒有收到確認消息，則返回false。
WaitForConfirms()的參數為自己規定的等待時間，並且有兩種重載。
13.2 單獨Confirm模式
在每次發送消息之后等待確認消息，對於生產者自身而言，是一種同步方法。
修改生產者WORK模式的代碼
for (int i = 0; i < 50; i++)
{
string message = "" + i;
var sendBytes = Encoding.UTF8.GetBytes($@"消息{message}");
channel.BasicPublish("", "隊列A", null, sendBytes);
Console.WriteLine($@" [x] Sent 消息{message}");
//等待Broker的確認消息
if (channel.WaitForConfirms(new TimeSpan(0, 0, 5)))
{
Console.WriteLine($@"接收到了確認信息");
Console.WriteLine();
}
Thread.Sleep(i * 10);
}
每發送一條消息后，生產者都會等待Borker的確認消息，再發送下一條消息。
優點：操作簡單
缺點：對確認消息的等待會阻止后續發送消息的操作，大大減慢發送速度
13.3 批量Confirm模式
為了改進單獨Confirm的缺點，避免為每一個發送的消息等待Broker的確認信息。可以使用批量Confirm的方法。
一次性發送N條消息，並在一次WaitForConfirms中獲取這N條消息的確認信息。
修改生產者Work模式的代碼
Console.WriteLine("生產者已啟動");
//定義每批發送消息的數量，每批發10條
int batchSize = 10;
//消息計數
int pubCount = 0;
for (int i = 0; i < 50; i++)
{
string message = "" + i;
var sendBytes = Encoding.UTF8.GetBytes($@"消息{message}");
channel.BasicPublish("", "隊列A", null, sendBytes);
Console.WriteLine($@" [x] Sent 消息{message}");
pubCount++;
if (pubCount == batchSize)
{
//等待Broker的確認消息
if (channel.WaitForConfirms(new TimeSpan(0, 0, 5)))
{
Console.WriteLine($@"接收到了確認信息");
Console.WriteLine();
}
//計數還原
pubCount = 0;
}
Thread.Sleep(i * 10);
}
定義每批消息的數量，並定義一個消息的計數器。
可以看到，每10個消息為一批，統一接受到Borker的Confirm消息。
優點：與單獨Confirm相比，有效提高了信息的吞吐量。
缺點：依然是同步操作，對確認信息的等待會阻塞后續操作。而且由於是按批獲取確認消息，如果出現問題無法得知具體是哪一條消息丟失。
13.4 異步Confirm模式
異步獲取每一條消息的確認信息。不需要使用WaitForConfirms()方法，只需要注冊兩個回調函數即可。
channel.BasicAcks：獲取的確認消息為ack時執行的回調函數
channel.BasicNacks：獲取的確認消息為nack時執行的回調函數
修改生產者WORK模式的代碼
/// <summary>
/// Work模式
/// </summary>
private static void WorkMode()
{
//創建連接工廠
ConnectionFactory factory = new ConnectionFactory
{
UserName = "guest",//用戶名
Password = "guest",//密碼
HostName = "localhost"//rabbitmq ip
};
var connection = factory.CreateConnection();
var channel = connection.CreateModel();
//開啟channel的Confirm模式
channel.ConfirmSelect();
//ACK回調函數(model指自己：生產者;ea:EventArgs參數)
channel.BasicAcks += (model, ea) =>
{
Console.WriteLine($@"接收到了的ACK,Tag為{ea.DeliveryTag}");
Console.WriteLine($@"Ea的Multiple:{ea.Multiple}");
Console.WriteLine();
};
//Nack回調函數
channel.BasicNacks += (model, ea) =>
{
Console.WriteLine($@"接收到了的NACK,Tag為{ea.DeliveryTag}");
Console.WriteLine($@"Ea的Multiple:{ea.Multiple}");
Console.WriteLine();
};
channel.QueueDeclare("隊列A", false, false, false, null);
Console.WriteLine("生產者已啟動");
for (int i = 0; i < 50; i++)
{
string message = "" + i;
var sendBytes = Encoding.UTF8.GetBytes($@"消息{message}");
channel.BasicPublish("", "隊列A", null, sendBytes);
Console.WriteLine($@" [x] Sent 消息{message}");
//等待Broker的確認消息
channel.WaitForConfirms(new TimeSpan(0, 0, 5));
Thread.Sleep(i * 10);
}
channel.Close();
connection.Close();
Console.ReadKey();
}
主要添加三部分，分別是：開啟Confirm，接受Confirm，編寫Confirm為ACK或NACK時的回調函數
兩個回調函數的參數都是相同的，其中ea有兩個參數：DeliveryTag和Multiple。
DeliveryTag（發送標簽）：是發送消息之前為消息打上的序號（從1開始）。
Multiple（bool）：本條Confirm對應一條還是多條。
優點：異步獲取信息，不會造成阻塞。並且可以利用TAG與 消息的關聯，在調用成功與失敗的回調函數中，多做一些事情。
缺點：稍微有些復雜，也沒那么復雜。