這里原來有一句話，觸犯啦天條，被閹割！！！！
首先不去討論我的日志組件怎么樣。因為有些日志需要走網絡，有的又不需要走網路，也是有性能與業務場景的多般變化在其中，就把他拋開，我們只談消息RabbitMQ。
那么什么是RabbitMQ，它是用來解決什么問題的，性能如何，又怎么用？我會在下面一一闡述，如有錯誤，不到之處，還望大家不吝賜教。
RabbitMQ簡介
必須一提的是rabbitmq是由LShift提供的一個消息隊列協議（AMQP)的開源實現，由以高性能、健壯以及可伸縮性出名的Erlang寫成（因此也是繼承了這些優點）。
百度百科對RabbitMQ闡述也非常明確，建議去看下，還有amqp協議。
RabbitMQ官網：http://www.rabbitmq.com/ 如果你要下載安裝，那么必須先把Erlang語言裝上。
RabbitMQ的.net客戶端，可以在nuget中輸入rabbitmq輕松獲得。
RabbitMQ與其他消息隊列的對比，早有仙人給寫出來。 Message Queue Shootout
這篇文章中的測試案例為：1百萬條1k的消息，每秒種的收發情況如下圖。
如果你安裝好啦，rabbitmq，他會提供一個操作監控頁面,頁面如下，他幾乎提供啦，對rabbitmq的所有操作，與監控，所以，你裝上后，自己多看看，多操作下。
RabbitMQ中的一些名詞闡述與消息從投遞到消費的整個過程
從上圖的標題中可以看到一些陌生的英文單詞，讓我們感覺一無所知，更無從操作，那么我給大家弄啦一個圖片大家可以看下，或許對您理解這些新鮮的單詞有所幫助。
看過這些名詞，之后，或許你還毫無頭緒，那么我把消息從生產到消費的整個流程給大家說一下，或許會更深入一點，其中Exchange，與Queue都是可以設置相關屬性，隊列的持久化，交換器類型制定。
Note：首先這個過程走分三個部分，1、客戶端（生產消息隊列），2、RabbitMQ服務端（負責路由規則的綁定與消息的分發），3、客戶端（消費消息隊列中的消息）
Note:由圖可以看出，一個消息可以走一次網絡卻被分發到不同的消息隊列中，然后被多個的客戶端消費，那么這個過程就是RabbitMQ的核心機制，RabbitMQ的路由類型與消費模式。
RabbitMQ中Exchange的類型
類型有4種，direct，fanout，topic，headers。其中headers不常用，本篇不做介紹，其他三種類型，會做詳細介紹。
那么這些類型是什么意思呢？就是Exchange與隊列進行綁定后，消息根據exchang的類型，按照不同的綁定規則分發消息到消息隊列中，可以是一個消息被分發給多個消息隊列，也可以是一個消息分發到一個消息隊列。具體請看下文。
介紹之初還要說下RoutingKey，這是個什么玩意呢？他是exchange與消息隊列綁定中的一個標識。有些路由類型會按照標識對應消息隊列，有些路由類型忽略routingkey。具體看下文。
1、Exchange類型direct
他是根據交換器名稱與routingkey來找隊列的。
Note:消息從client發出，傳送給交換器ChangeA，RoutingKey為routingkey.ZLH,那么不管你發送給Queue1，還是Queue2一個消息都會保存在Queue1，Queue2，Queue3，三個隊列中。這就是交換器的direct類型的路由規則。只要找到路由器與routingkey綁定的隊列，那么他有多少隊列，他就分發給多少隊列。
2、Exchange類型fanout
這個類型忽略Routingkey，他為廣播模式。
Note:消息從客戶端發出，只要queue與exchange有綁定，那么他不管你的Routingkey是什么他都會將消息分發給所有與該exchang綁定的隊列中。
3、Exchange類型topic
這個類型的路由規則如果你掌握啦，那是相當的好用，與靈活。他是根據RoutingKey的設置，來做匹配的，其中這里還有兩個通配符為：
*，代表任意的一個詞。例如topic.zlh.*，他能夠匹配到，topic.zlh.one ,topic.zlh.two ,topic.zlh.abc, ....
#，代表任意多個詞。例如topic.#，他能夠匹配到，topic.zlh.one ,topic.zlh.two ,topic.zlh.abc, ....
Note：這個圖看上去很亂，但是他是根據匹配符做匹配的，這里我建議你自己做下消息隊列的具體操作。
具體操作如下
public static void Producer(int value)
{
try
{
var qName = "lhtest1";
var exchangeName = "fanoutchange1";
var exchangeType = "fanout";//topic、fanout
var routingKey = "*";
var uri = new Uri("amqp://192.168.10.121:5672/");
var factory = new ConnectionFactory
{
UserName = "123",
Password = "123",
RequestedHeartbeat = 0,
Endpoint = new AmqpTcpEndpoint(uri)
};
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
//設置交換器的類型
channel.ExchangeDeclare(exchangeName, exchangeType);
//聲明一個隊列，設置隊列是否持久化，排他性，與自動刪除
channel.QueueDeclare(qName, true, false, false, null);
//綁定消息隊列，交換器，routingkey
channel.QueueBind(qName, exchangeName, routingKey);
var properties = channel.CreateBasicProperties();
//隊列持久化
properties.Persistent = true;
var m = new QMessage(DateTime.Now, value+"");
var body = Encoding.UTF8.GetBytes(DoJson.ModelToJson<QMessage>(m));
//發送信息
channel.BasicPublish(exchangeName, routingKey, properties, body);
}
}
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
}
消息隊列的消費與消息確認Ack
1、消息隊列的消費
Note:如果一個消息隊列中有大量消息等待操作時，我們可以用多個客戶端來處理消息，這里的分發機制是采用負載均衡算法中的輪詢。第一個消息給A，下一個消息給B，下下一個消息給A，下下下一個消息給B......以此類推。
2、為啦保證消息的安全性，保證此消息被正確處理后才能在服務端的消息隊列中刪除。那么rabbitmq提供啦ack應答機制，來實現這一功能。
ack應答有兩種方式：1、自動應答，2、手動應答。具體實現如下。
public static void Consumer()
{
try
{
var qName = "lhtest1";
var exchangeName = "fanoutchange1";
var exchangeType = "fanout";//topic、fanout
var routingKey = "*";
var uri = new Uri("amqp://192.168.10.121:5672/");
var factory = new ConnectionFactory
{
UserName = "123",
Password = "123",
RequestedHeartbeat = 0,
Endpoint = new AmqpTcpEndpoint(uri)
};
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchangeName, exchangeType);
channel.QueueDeclare(qName, true, false, false, null);
channel.QueueBind(qName, exchangeName, routingKey);
//定義這個隊列的消費者
QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
//false為手動應答，true為自動應答
channel.BasicConsume(qName, false, consumer);
while (true)
{
BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
byte[] bytes = ea.Body;
var messageStr = Encoding.UTF8.GetString(bytes);
var message = DoJson.JsonToModel<QMessage>(messageStr);
Console.WriteLine("Receive a Message, DateTime:" + message.DateTime.ToString("yyyy-MM-dd HH:mm:ss") + " Title:" + message.Title);
//如果是自動應答，下下面這句代碼不用寫啦。
if ((Convert.ToInt32(message.Title) % 2) == 1)
{
channel.BasicAck(ea.DeliveryTag, false);
}
}
}
}
}
RabbitMQ持久化機制
核心代碼：
channel.queueDeclare(queue_name, durable, false, false, null); //聲明消息隊列，且為可持久化的
String message="Hello world"+Math.random();
//將隊列設置為持久化之后，還需要將消息也設為可持久化的，MessageProperties.PERSISTENT_TEXT_PLAIN
channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
消息什么時候需要持久化？
根據 官方博文 的介紹，RabbitMQ在兩種情況下會將消息寫入磁盤：
- 消息本身在publish的時候就要求消息寫入磁盤；
- 內存緊張，需要將部分內存中的消息轉移到磁盤；
消息什么時候會刷到磁盤？
- 寫入文件前會有一個Buffer，大小為1M（1048576），數據在寫入文件時，首先會寫入到這個Buffer，如果Buffer已滿，則會將Buffer寫入到文件（未必刷到磁盤）；
- 有個固定的刷盤時間：25ms，也就是不管Buffer滿不滿，每隔25ms，Buffer里的數據及未刷新到磁盤的文件內容必定會刷到磁盤；
- 每次消息寫入后，如果沒有后續寫入請求，則會直接將已寫入的消息刷到磁盤：使用Erlang的receive x after 0來實現，只要進程的信箱里沒有消息，則產生一個timeout消息，而timeout會觸發刷盤操作。
消息在磁盤文件中的格式
消息保存於$MNESIA/msg_store_persistent/x.rdq文件中，其中x為數字編號，從1開始，每個文件最大為16M（16777216），超過這個大小會生成新的文件，文件編號加1。消息以以下格式存在於文件中：
<<Size:64, MsgId:16/binary, MsgBody>>
MsgId為RabbitMQ通過rabbit_guid:gen()每一個消息生成的GUID，MsgBody會包含消息對應的exchange，routing_keys，消息的內容，消息對應的協議版本，消息內容格式（二進制還是其它）等等。
文件何時刪除？
當所有文件中的垃圾消息（已經被刪除的消息）比例大於閾值（GARBAGE_FRACTION = 0.5）時，會觸發文件合並操作（至少有三個文件存在的情況下），以提高磁盤利用率。
publish消息時寫入內容，ack消息時刪除內容（更新該文件的有用數據大小），當一個文件的有用數據等於0時，刪除該文件。
消息索引什么時候需要持久化？
索引的持久化與消息的持久化類似，也是在兩種情況下需要寫入到磁盤中：要么本身需要持久化，要么因為內存緊張，需要釋放部分內存。
消息索引什么時候會刷到磁盤？
- 有個固定的刷盤時間：25ms，索引文件內容必定會刷到磁盤；
- 每次消息（及索引）寫入后，如果沒有后續寫入請求，則會直接將已寫入的索引刷到磁盤，實現上與消息的timeout刷盤一致。
RabbitMQ(二)隊列與消息的持久化
當有多個消費者同時收取消息，且每個消費者在接收消息的同時，還要做其它的事情，且會消耗很長的時間，在此過程中可能會出現一些意外，比如消息接收到一半的時候，一個消費者宕掉了，這時候就要使用消息接收確認機制，可以讓其它的消費者再次執行剛才宕掉的消費者沒有完成的事情。另外，在默認情況下，我們創建的消息隊列以及存放在隊列里面的消息，都是非持久化的，也就是說當RabbitMQ宕掉了或者是重啟了，創建的消息隊列以及消息都不會保存，為了解決這種情況，保證消息傳輸的可靠性，我們可以使用RabbitMQ提供的消息隊列的持久化機制。
生產者：
2 import com.rabbitmq.client.Connection;
3 import com.rabbitmq.client.Channel;
4 import com.rabbitmq.client.MessageProperties;
5 public class ClientSend1 {
6 public static final String queue_name="my_queue";
7 public static final boolean durable=true; //消息隊列持久化
8 public static void main(String[] args)
9 throws java.io.IOException{
10 ConnectionFactory factory=new ConnectionFactory(); //創建連接工廠
11 factory.setHost("localhost");
12 factory.setVirtualHost("my_mq");
13 factory.setUsername("zhxia");
14 factory.setPassword("123456");
15 Connection connection=factory.newConnection(); //創建連接
16 Channel channel=connection.createChannel();//創建信道
17 channel.queueDeclare(queue_name, durable, false, false, null); //聲明消息隊列，且為可持久化的
18 String message="Hello world"+Math.random();
19 //將隊列設置為持久化之后，還需要將消息也設為可持久化的，MessageProperties.PERSISTENT_TEXT_PLAIN
20 channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
21 System.out.println("Send message:"+message);
22 channel.close();
23 connection.close();
24 }
25
26 }
說明：
行17 和行20 需要同時設置，也就是將隊列設置為持久化之后，還需要將發送的消息也要設置為持久化才能保證隊列和消息一直存在
消費者：
2 import com.rabbitmq.client.Connection;
3 import com.rabbitmq.client.Channel;
4 import com.rabbitmq.client.QueueingConsumer;
5 public class ClientReceive1 {
6 public static final String queue_name="my_queue";
7 public static final boolean autoAck=false;
8 public static final boolean durable=true;
9 public static void main(String[] args)
10 throws java.io.IOException,java.lang.InterruptedException{
11 ConnectionFactory factory=new ConnectionFactory();
12 factory.setHost("localhost");
13 factory.setVirtualHost("my_mq");
14 factory.setUsername("zhxia");
15 factory.setPassword("123456");
16 Connection connection=factory.newConnection();
17 Channel channel=connection.createChannel();
18 channel.queueDeclare(queue_name, durable, false, false, null);
19 System.out.println("Wait for message");
20 channel.basicQos(1); //消息分發處理
21 QueueingConsumer consumer=new QueueingConsumer(channel);
22 channel.basicConsume(queue_name, autoAck, consumer);
23 while(true){
24 Thread.sleep(500);
25 QueueingConsumer.Delivery deliver=consumer.nextDelivery();
26 String message=new String(deliver.getBody());
27 System.out.println("Message received:"+message);
28 channel.basicAck(deliver.getEnvelope().getDeliveryTag(), false);
29 }
30 }
31 }
說明：
行22： 設置RabbitMQ調度分發消息的方式，也就是告訴RabbitMQ每次只給消費者處理一條消息，也就是等待消費者處理完並且已經對剛才處理的消息進行確認之后， 才發送下一條消息，防止消費者太過於忙碌。如下圖所示：
