RabbitMQ必備核心知識（精選）
現在很多知名的網際網路公司都有用到RabbitMQ，其效能，可擴充套件性讓很多大公司青睞於使用它，不過想要完全使用好RabbitMQ需要掌握其核心的一些概念，這裡就說說掌握RabbitMQ所需的必要知識
生產者與消費者
生產者： 建立訊息，然後傳送到代理伺服器(RabbitMQ)的程式
消費者：連線到代理伺服器，並訂閱到佇列上接收訊息
訊息流程
AMQP協議規定，AMQP訊息必須有三部分，交換機，佇列和繫結。生產者把訊息傳送到交換機，交換機與佇列的繫結關係決定了訊息如何路由到特定的佇列，最終被消費者接收。
Note ： 訊息是不能直接到達佇列(Queue)的
交換機
訊息實際上投遞到的是交換機，具體路由到那個佇列由交換機根據路由鍵(routing key)完成。
當你發訊息到代理伺服器時，即便路由鍵是空的，RabbitMQ也會將其和使用的路由鍵進行匹配。如果路由的訊息不匹配任何繫結模式，訊息將會進入黑洞。
交換機在佇列與訊息中間起到了中間層的作用，有了交換機我們可以實現更靈活的功能，RabbitMQ中有三種常用的交換機型別：
direct: 如果路由鍵匹配，訊息就投遞到對應的佇列
fanout：投遞訊息給所有繫結在當前交換機上面的佇列
topic：允許實現有趣的訊息通訊場景，使得5不同源頭的訊息能夠達到同一個佇列。topic佇列名稱有兩個特殊的關鍵字。
* 可以替換一個單詞
# 可以替換所有的單詞
可以理解，direct為1v1, fanout為1v所有，topic比較靈活，可以1v任意。
虛擬主機
每一個虛擬主機(vhost)相當於mini版的RabbitMQ伺服器，擁有自己的佇列，交換機和繫結，許可權… 這使得一個RabbitMQ服務眾多的應用程式，而不會互相沖突。
rabbitMQ預設的虛擬主機為: “/” ，一般我們在建立Rabbit的使用者時會再給使用者分配一個虛擬主機。
操作虛擬主機，除了命令列之外還有一個web管理頁面
#建立虛擬主機 rabbitmqctl add vhost [vhost_name] #刪除虛擬主機 rabbitmqctl delete vhost [vhost_name] #列出虛擬主機 rabbitmqctl list_vhosts
訊息投遞策略
預設情況下RabbitMQ的佇列和交換機在RabbitMQ伺服器重啟之後會消失，原因在於佇列和交換機的durable屬性，該屬性預設情況下為false.
能從AMQP伺服器崩潰中恢復的訊息稱為持久化訊息，如果想要從崩潰中恢復那麼訊息必須
投遞模式設定2，來標記訊息為持久化
傳送到持久化的交換機
到到持久化的佇列
缺點：訊息寫入磁碟效能差很多。除非特別關鍵的訊息會使用
關鍵API
以上都是概念性的內容，實際我們還是要通過程式設計來實現我們的目的，RabbitMQ的客戶端api提供了很多功能，通過看程式碼，來了解它的強大之處。
基本步驟之前的RabbitMQ快速入門已經提過了，Channel類是關鍵的部分：包含了很多我們想要的功能
訊息確認
生成端可以新增監聽事件：
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.err.println("-------no ack!-----------");
}
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.err.println("-------ack!-----------");
}
});
消費端可以確認訊息狀態：
public class MyConsumer extends DefaultConsumer {
private Channel channel ;
public MyConsumer(Channel channel) {
super(channel);
this.channel = channel;
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.err.println("-----------consume message----------");
System.err.println("body: " + new String(body));
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
if((Integer)properties.getHeaders().get("num") == 0) {
channel.basicNack(envelope.getDeliveryTag(), false, true);
} else {
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
}
channel.basicAck與basicNack最後一個引數指定訊息是否重回佇列。
監聽不可達訊息
我們的訊息生產者通過指定交換機和路由鍵來把訊息送到佇列中，但有時候指定的路由鍵不存在，或者交換機不存在，那麼訊息就會return，我們可以通過新增return listener來實現：
channel.addReturnListener(new ReturnListener() {
@Override
public void handleReturn(int replyCode, String replyText, String exchange,
String routingKey, BasicProperties properties, byte[] body) throws IOException {
System.err.println("---------handle return----------");
System.err.println("replyCode: " + replyCode);
System.err.println("replyText: " + replyText);
System.err.println("exchange: " + exchange);
System.err.println("routingKey: " + routingKey);
System.err.println("properties: " + properties);
System.err.println("body: " + new String(body));
}
});
channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
在basicPublish中的Mandatory要設定為true才會生效，否則broker會刪除該訊息
消費端限流
假設MQ伺服器上面囤積了成千上萬條的訊息的時候，這個時候突然連線消費端，那麼巨量的訊息全部推過來，但是客戶端無法一次性處理這麼多的資料。
在高併發的時候，瞬間產生的流量很大，訊息很大，而MQ有個重要的作用就是限流，限流則是消費端做的。
RabbitMQ提供了一種Qos(服務質量保證)功能，即在非自動確認訊息的前提下，在一定數量的訊息未被消費前，不進行消費新的訊息。
// prefetchSize訊息的限制大小，一般設定為0，在生產端限制 // prefetchCount 我們一次最多消費多少條訊息，一般設定為1 // global，一般設定為false，在消費端進行限制 channel.basicQos(int prefetchSize, int prefetchCount, boolean global) // 使用 channel.basicQos(0, 1, false); channel.basicConsume(queueName, false, new MyConsumer(channel));
Note ： autoAck設定為false, 一定要手工簽收訊息
死信佇列(DLX)
當訊息在佇列中變成死信，沒有消費者進行消費的時候，訊息可能會被重新發布到另外一個佇列中，這個佇列就是死信佇列。
以下情況會導致訊息進入死信佇列：
basic.reject/basic.nack 並且 requeue為false(不重回佇列)的時候，訊息就是死信
訊息TTL過期
佇列達到最大的長度
死信佇列也是正常的Exchange，和一般的Exchange沒什麼區別，不過要做一點操作。
設定死信佇列包括：
設定Exchange(dlx.exchange名稱隨意),設定Queue(dlx.queue),設定RoutingKey(#)
建立正常的交換機，佇列，繫結，只不過加上一個引數 arguments.put(“x-dead-letter-exchange”,“dlx.exchange”)
// 這就是一個普通的交換機 和 佇列 以及路由 String exchangeName = "test_dlx_exchange"; String routingKey = "dlx.#"; String queueName = "test_dlx_queue"; channel.exchangeDeclare(exchangeName, "topic", true, false, null); Map agruments = new HashMap(); agruments.put("x-dead-letter-exchange", "dlx.exchange"); //這個agruments屬性，要設定到宣告佇列上 channel.queueDeclare(queueName, true, false, false, agruments); channel.queueBind(queueName, exchangeName, routingKey); //要進行死信佇列的宣告: channel.exchangeDeclare("dlx.exchange", "topic", true, false, null); channel.queueDeclare("dlx.queue", true, false, false, null); channel.queueBind("dlx.queue", "dlx.exchange", "#");
寫在最後：歡迎留言討論，加關注，持續更新！