消息隊列模型深刻理解
消息隊列模型深刻理解
RabbitMQ提供了6種消息模型，可是第6種實際上是RPC，並非MQ，所以不予學習。那麼也就剩下5種。java
可是其實三、四、5這三種都屬於訂閱模型，只不過進行路由的方式不一樣。面試
2.1 基本消息模型
RabbitMQ是一個消息代理：它接受和轉發消息。 你能夠把它想象成一個郵局：當你把郵件放在郵箱裏時，你能夠肯定郵差先生最終會把郵件發送給你的收件人。 在這個比喻中，RabbitMQ是郵政信箱，郵局和郵遞員。spring
RabbitMQ與郵局的主要區別是它不處理紙張，而是接受，存儲和轉發數據消息的二進制數據塊。數據庫
P（producer/ publisher）：生產者，一個發送消息的用戶應用程序。數組
C（consumer）：消費者，消費和接收有相似的意思，消費者是一個主要用來等待接收消息的用戶應用程序app
隊列（紅色區域）：rabbitmq內部相似於郵箱的一個概念。雖然消息流經rabbitmq和你的應用程序，可是它們只能存儲在隊列中。隊列只受主機的內存和磁盤限制，實質上是一個大的消息緩衝區。許多生產者能夠發送消息到一個隊列，許多消費者能夠嘗試從一個隊列接收數據。框架
總之：異步
生產者將消息發送到隊列，消費者從隊列中獲取消息，隊列是存儲消息的緩衝區。ide
咱們將用Java編寫兩個程序;發送單個消息的生產者，以及接收消息並將其打印出來的消費者。咱們將詳細介紹Java API中的一些細節，這是一個消息傳遞的「Hello World」。spring-boot
咱們將調用咱們的消息發佈者（發送者）Send和咱們的消息消費者（接收者）Recv。發佈者將鏈接到RabbitMQ，發送一條消息，而後退出。
2.1.1.生產者發送消息
public class Send {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接以及mq通道
Connection connection = ConnectionUtil.getConnection();
// 從鏈接中建立通道，這是完成大部分API的地方。
Channel channel = connection.createChannel();
// 聲明（建立）隊列，必須聲明隊列纔可以發送消息，咱們能夠把消息發送到隊列中。
// 聲明一個隊列是冪等的 - 只有當它不存在時纔會被建立
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 消息內容
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
//關閉通道和鏈接
channel.close();
connection.close();
}
}
控制檯：
2.1.2.管理工具中查看消息
進入隊列頁面，能夠看到新建了一個隊列：simple_queue
點擊隊列名稱，進入詳情頁，能夠查看消息：
在控制檯查看消息並不會將消息消費，因此消息還在。
2.1.3.消費者獲取消息
public class Recv {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 建立通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [x] received : " + msg + "!");
}
};
// 監聽隊列，第二個參數：是否自動進行消息確認。
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
控制檯：
這個時候，隊列中的消息就沒了：
咱們發現，消費者已經獲取了消息，可是程序沒有中止，一直在監聽隊列中是否有新的消息。一旦有新的消息進入隊列，就會當即打印.
2.1.4.消息確認機制（ACK）
經過剛纔的案例能夠看出，消息一旦被消費者接收，隊列中的消息就會被刪除。
那麼問題來了：RabbitMQ怎麼知道消息被接收了呢？
若是消費者領取消息後，還沒執行操做就掛掉了呢？或者拋出了異常？消息消費失敗，可是RabbitMQ無從得知，這樣消息就丟失了！
所以，RabbitMQ有一個ACK機制。當消費者獲取消息後，會向RabbitMQ發送回執ACK，告知消息已經被接收。不過這種回執ACK分兩種狀況：
- 自動ACK：消息一旦被接收，消費者自動發送ACK
- 手動ACK：消息接收後，不會發送ACK，須要手動調用
你們以爲哪一種更好呢？
這須要看消息的重要性：
- 若是消息不過重要，丟失也沒有影響，那麼自動ACK會比較方便
- 若是消息很是重要，不容丟失。那麼最好在消費完成後手動ACK，不然接收消息後就自動ACK，RabbitMQ就會把消息從隊列中刪除。若是此時消費者宕機，那麼消息就丟失了。
咱們以前的測試都是自動ACK的，若是要手動ACK，須要改動咱們的代碼：
public class Recv2 {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 建立通道
final Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [x] received : " + msg + "!");
// 手動進行ACK
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 監聽隊列，第二個參數false，手動進行ACK
channel.basicConsume(QUEUE_NAME, false, consumer);
}
}
注意到最後一行代碼：
channel.basicConsume(QUEUE_NAME, false, consumer);
若是第二個參數爲true，則會自動進行ACK；若是爲false，則須要手動ACK。方法的聲明：
2.1.4.1.自動ACK存在的問題
修改消費者，添加異常，以下：
生產者不作任何修改，直接運行，消息發送成功：
運行消費者，程序拋出異常。可是消息依然被消費：
管理界面：
2.1.4.2.演示手動ACK
修改消費者，把自動改爲手動（去掉以前製造的異常）
生產者不變，再次運行：
運行消費者
可是，查看管理界面，發現：
停掉消費者的程序，發現：
這是由於雖然咱們設置了手動ACK，可是代碼中並無進行消息確認！因此消息並未被真正消費掉。
當咱們關掉這個消費者，消息的狀態再次稱爲Ready
修改代碼手動ACK：
執行：
消息消費成功！
2.2.work消息模型
工做隊列或者競爭消費者模式
在第一篇教程中，咱們編寫了一個程序，從一個命名隊列中發送並接受消息。在這裏，咱們將建立一個工做隊列，在多個工做者之間分配耗時任務。
工做隊列，又稱任務隊列。主要思想就是避免執行資源密集型任務時，必須等待它執行完成。相反咱們稍後完成任務，咱們將任務封裝爲消息並將其發送到隊列。 在後臺運行的工做進程將獲取任務並最終執行做業。當你運行許多工人時，任務將在他們之間共享，可是一個消息只能被一個消費者獲取。
這個概念在Web應用程序中特別有用，由於在短的HTTP請求窗口中沒法處理複雜的任務。
接下來咱們來模擬這個流程：
P：生產者：任務的發佈者
C1：消費者，領取任務而且完成任務，假設完成速度較快
C2：消費者2：領取任務並完成任務，假設完成速度慢
面試題：避免消息堆積？
1） 採用workqueue，多個消費者監聽同一隊列。
2）接收到消息之後，而是經過線程池，異步消費。
2.2.1.生產者
生產者與案例1中的幾乎同樣：
public class Send {
private final static String QUEUE_NAME = "test_work_queue";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 循環發佈任務
for (int i = 0; i < 50; i++) {
// 消息內容
String message = "task .. " + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
Thread.sleep(i * 2);
}
// 關閉通道和鏈接
channel.close();
connection.close();
}
}
不過這裏咱們是循環發送50條消息。
2.2.2.消費者1
2.2.3.消費者2
與消費者1基本相似，就是沒有設置消費耗時時間。
這裏是模擬有些消費者快，有些比較慢。
接下來，兩個消費者一同啓動，而後發送50條消息：
能夠發現，兩個消費者各自消費了25條消息，並且各不相同，這就實現了任務的分發。
2.2.4.能者多勞
剛纔的實現有問題嗎？
- 消費者1比消費者2的效率要低，一次任務的耗時較長
- 然而兩人最終消費的消息數量是同樣的
- 消費者2大量時間處於空閒狀態，消費者1一直忙碌
如今的狀態屬因而把任務平均分配，正確的作法應該是消費越快的人，消費的越多。
怎麼實現呢？
咱們可使用basicQos方法和prefetchCount = 1設置。 這告訴RabbitMQ一次不要向工做人員發送多於一條消息。 或者換句話說，不要向工做人員發送新消息，直到它處理並確認了前一個消息。 相反，它會將其分派給不是仍然忙碌的下一個工做人員。
再次測試：
2.3.訂閱模型分類
在以前的模式中，咱們建立了一個工做隊列。 工做隊列背後的假設是：每一個任務只被傳遞給一個工做人員。 在這一部分，咱們將作一些徹底不一樣的事情 - 咱們將會傳遞一個信息給多個消費者。 這種模式被稱爲「發佈/訂閱」。
訂閱模型示意圖：
解讀：
一、1個生產者，多個消費者
二、每個消費者都有本身的一個隊列
三、生產者沒有將消息直接發送到隊列，而是發送到了交換機
四、每一個隊列都要綁定到交換機
五、生產者發送的消息，通過交換機到達隊列，實現一個消息被多個消費者獲取的目的
X（Exchanges）：交換機一方面：接收生產者發送的消息。另外一方面：知道如何處理消息，例如遞交給某個特別隊列、遞交給全部隊列、或是將消息丟棄。到底如何操做，取決於Exchange的類型。
Exchange類型有如下幾種：
Fanout：廣播，將消息交給全部綁定到交換機的隊列
Direct：定向，把消息交給符合指定routing key 的隊列
Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列
咱們這裏先學習
Fanout：即廣播模式
Exchange（交換機）只負責轉發消息，不具有存儲消息的能力，所以若是沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那麼消息會丟失！
2.4.訂閱模型-Fanout
Fanout，也稱爲廣播。
流程圖：
在廣播模式下，消息發送流程是這樣的：
- 1） 能夠有多個消費者
- 2） 每一個消費者有本身的queue（隊列）
- 3） 每一個隊列都要綁定到Exchange（交換機）
- 4） 生產者發送的消息，只能發送到交換機，交換機來決定要發給哪一個隊列，生產者沒法決定。
- 5） 交換機把消息發送給綁定過的全部隊列
- 6） 隊列的消費者都能拿到消息。實現一條消息被多個消費者消費
2.4.1.生產者
兩個變化：
- 1） 聲明Exchange，再也不聲明Queue
- 2） 發送消息到Exchange，再也不發送到Queue
public class Send {
private final static String EXCHANGE_NAME = "fanout_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明exchange，指定類型爲fanout
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 消息內容
String message = "Hello everyone";
// 發佈消息到Exchange
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [生產者] Sent '" + message + "'");
channel.close();
connection.close();
}
}
2.4.2.消費者1
public class Recv {
private final static String QUEUE_NAME = "fanout_exchange_queue_1";
private final static String EXCHANGE_NAME = "fanout_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者1] received : " + msg + "!");
}
};
// 監聽隊列，自動返回完成
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
要注意代碼中：隊列須要和交換機綁定
2.4.3.消費者2
public class Recv2 {
private final static String QUEUE_NAME = "fanout_exchange_queue_2";
private final static String EXCHANGE_NAME = "fanout_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，手動返回完成
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
2.4.4.測試
咱們運行兩個消費者，而後發送1條消息：
2.5.訂閱模型-Direct
有選擇性的接收消息
在訂閱模式中，生產者發佈消息，全部消費者均可以獲取全部消息。
在路由模式中，咱們將添加一個功能 - 咱們將只能訂閱一部分消息。 例如，咱們只能將重要的錯誤消息引導到日誌文件（以節省磁盤空間），同時仍然可以在控制檯上打印全部日誌消息。
可是，在某些場景下，咱們但願不一樣的消息被不一樣的隊列消費。這時就要用到Direct類型的Exchange。
在Direct模型下，隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）
消息的發送方在向Exchange發送消息時，也必須指定消息的routing key。
P：生產者，向Exchange發送消息，發送消息時，會指定一個routing key。
X：Exchange（交換機），接收生產者的消息，而後把消息遞交給 與routing key徹底匹配的隊列
C1：消費者，其所在隊列指定了須要routing key 爲 error 的消息
C2：消費者，其所在隊列指定了須要routing key 爲 info、error、warning 的消息
2.5.1.生產者
此處咱們模擬商品的增刪改，發送消息的RoutingKey分別是：insert、update、delete
public class Send {
private final static String EXCHANGE_NAME = "direct_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明exchange，指定類型爲direct
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
// 消息內容
String message = "商品新增了， id = 1001";
// 發送消息，而且指定routing key 爲：insert ,表明新增商品
channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
System.out.println(" [商品服務：] Sent '" + message + "'");
channel.close();
connection.close();
}
}
2.5.2.消費者1
咱們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品。
public class Recv {
private final static String QUEUE_NAME = "direct_exchange_queue_1";
private final static String EXCHANGE_NAME = "direct_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定須要訂閱的routing key。假設此處須要update和delete消息
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者1] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
2.5.3.消費者2
咱們此處假設消費者2接收全部類型的消息：新增商品，更新商品和刪除商品。
public class Recv2 {
private final static String QUEUE_NAME = "direct_exchange_queue_2";
private final static String EXCHANGE_NAME = "direct_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定須要訂閱的routing key。訂閱 insert、update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
2.5.4.測試
咱們分別發送增、刪、改的RoutingKey，發現結果：
2.6.訂閱模型-Topic
Topic類型的Exchange與Direct相比，都是能夠根據RoutingKey把消息路由到不一樣的隊列。只不過Topic類型Exchange可讓隊列在綁定Routing key 的時候使用通配符！
Routingkey 通常都是有一個或多個單詞組成，多個單詞之間以」.」分割，例如： item.insert
通配符規則：
#：匹配一個或多個詞
*：匹配很少很多剛好1個詞
舉例：
audit.#：可以匹配audit.irs.corporate 或者 audit.irs
audit.*：只能匹配audit.irs
在這個例子中，咱們將發送全部描述動物的消息。消息將使用由三個字（兩個點）組成的routing key發送。路由關鍵字中的第一個單詞將描述速度，第二個顏色和第三個種類：「
咱們建立了三個綁定：Q1綁定了綁定鍵「* .orange.」，Q2綁定了「.*.rabbit」和「lazy.＃」。
Q1匹配全部的橙色動物。
Q2匹配關於兔子以及懶惰動物的消息。
練習，生產者發送以下消息，會進入那個隊列：
quick.orange.rabbit à Q1 Q2
lazy.orange.elephant à Q1 Q2
quick.orange.fox à Q1
lazy.pink.rabbit à Q2
quick.brown.fox à 不匹配任意隊列，被丟棄
quick.orange.male.rabbit à
orange à
2.6.1.生產者
使用topic類型的Exchange，發送消息的routing key有3種： item.isnert、item.update、item.delete：
public class Send {
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明exchange，指定類型爲topic
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// 消息內容
String message = "新增商品 : id = 1001";
// 發送消息，而且指定routing key 爲：insert ,表明新增商品
channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
System.out.println(" [商品服務：] Sent '" + message + "'");
channel.close();
connection.close();
}
}
2.6.2.消費者1
咱們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品
public class Recv {
private final static String QUEUE_NAME = "topic_exchange_queue_1";
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定須要訂閱的routing key。須要 update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者1] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
2.6.3.消費者2
咱們此處假設消費者2接收全部類型的消息：新增商品，更新商品和刪除商品。
/**
* 消費者2
*/
public class Recv2 {
private final static String QUEUE_NAME = "topic_exchange_queue_2";
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到鏈接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定須要訂閱的routing key。訂閱 insert、update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，而且處理，這個方法相似事件監聽，若是有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
2.7.持久化
如何避免消息丟失？
1） 消費者的ACK機制。能夠防止消費者丟失消息。
2） 可是，若是在消費者消費以前，MQ就宕機了，消息就沒了。
是能夠將消息進行持久化呢？
要將消息持久化，前提是：隊列、Exchange都持久化
2.7.1.交換機持久化
2.7.2.隊列持久化
2.7.3.消息持久化
解決消息丟失？
- ack（消費者確認）
- 持久化（保證消息隊列可靠）
- 發送消息前，將消息持久化到數據庫，並記錄消息狀態（可靠消息服務）
- 生產者確認（publisher confirm）
冪等性（同一接口被重複執行，其結果一致）
3.Spring AMQP
2.2.依賴和配置
添加AMQP的啓動器：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
在application.yml中添加RabbitMQ地址：
spring:
rabbitmq:
host: 192.168.56.101
username: leyou
password: leyou
virtual-host: /leyou
2.3.監聽者
在SpringAmqp中，對消息的消費者進行了封裝和抽象，一個普通的JavaBean中的普通方法，只要經過簡單的註解，就能夠成爲一個消費者。
@Component
public class Listener {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "spring.test.queue", durable = "true"),
exchange = @Exchange(
value = "spring.test.exchange",
ignoreDeclarationExceptions = "true",
type = ExchangeTypes.TOPIC
),
key = {"#.#"}))
public void listen(String msg){
System.out.println("接收到消息：" + msg);
}
}
@Componet：類上的註解，註冊到Spring容器@RabbitListener：方法上的註解，聲明這個方法是一個消費者方法，須要指定下面的屬性：bindings：指定綁定關係，能夠有多個。值是@QueueBinding的數組。@QueueBinding包含下面屬性：value：這個消費者關聯的隊列。值是@Queue，表明一個隊列exchange：隊列所綁定的交換機，值是@Exchange類型key：隊列和交換機綁定的RoutingKey
相似listen這樣的方法在一個類中能夠寫多個，就表明多個消費者。
2.4.AmqpTemplate
Spring最擅長的事情就是封裝，把他人的框架進行封裝和整合。
Spring爲AMQP提供了統一的消息處理模板：AmqpTemplate，很是方便的發送消息，其發送方法：
紅框圈起來的是比較經常使用的3個方法，分別是：
- 指定交換機、RoutingKey和消息體
- 指定消息
- 指定RoutingKey和消息，會向默認的交換機發送消息
2.5.測試代碼
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {
@Autowired
private AmqpTemplate amqpTemplate;
@Test
public void testSend() throws InterruptedException {
String msg = "hello, Spring boot amqp";
this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg);
// 等待10秒後再結束
Thread.sleep(10000);
}
}
運行後查看日誌：
- 1. 消息隊列深刻解析
- 2. RabbitMQ系列三 （深刻消息隊列）
- 3. RabbitMQ 消息隊列之隊列模型
- 4. 深刻淺出 消息隊列 ActiveMQ
- 5. 深刻理解TAILQ隊列
- 6. 消息隊列的理解
- 7. 消息隊列: 快速理解消息隊列作用
- 8. 淺談異步消息隊列模型
- 9. [Android] android的消息隊列模型
- 10. Android 的消息隊列模型
- 更多相關文章...
- • C# 隊列（Queue） - C#教程
- • HTTP 消息結構 - HTTP 教程
- • 委託模式
- • Kotlin學習（二）基本類型
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 消息隊列深刻解析
- 2. RabbitMQ系列三 （深刻消息隊列）
- 3. RabbitMQ 消息隊列之隊列模型
- 4. 深刻淺出 消息隊列 ActiveMQ
- 5. 深刻理解TAILQ隊列
- 6. 消息隊列的理解
- 7. 消息隊列: 快速理解消息隊列作用
- 8. 淺談異步消息隊列模型
- 9. [Android] android的消息隊列模型
- 10. Android 的消息隊列模型