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 測驗
兩個專案一起啟動之后
在生產者對應的控制臺輸入文字后，添加到訊息佇列中，由消費者進行消費，顯示在消費者控制臺上，
參考檔案:https://zhuanlan.zhihu.com/p/143521328
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獲得并消費，
最終，這條訊息就被消費了兩次，
如果我們不人為保證訊息的冪等性，資料就會出錯，
關于訊息的等冪性，請看另一篇文章：
https://www.cnblogs.com/soraxtube/p/14816681.html
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與 訊息的關聯，在呼叫成功與失敗的回呼函式中，多做一些事情，
缺點：稍微有些復雜，也沒那么復雜，
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/285364.html
標籤：.NET Core