瞭解一下RabbitMQ

RabbitMQ · 發表 2018-10-31 00:18:00

摘要： RabbitMQ概述 RabbitMQ是遵從AMQP協議的通訊協議都設計到報文互動，換句話說RabbitMQ就是AMQP協議的Erlang的實現。 AMQP說到底還是一個通訊協議從low-level層面舉例來說，AMQP本身是應用層的協議，其填充於TCP協議的資料部分。從hig...

RabbitMQ概述

RabbitMQ是遵從AMQP協議的通訊協議都設計到報文互動，換句話說RabbitMQ就是AMQP協議的Erlang的實現。

AMQP說到底還是一個通訊協議從low-level層面舉例來說，AMQP本身是應用層的協議，其填充於TCP協議的資料部分。

從high-level層面來說，AMQP是通過協議命令進行互動的。命令類似HTTP中的方法(GET PUT POST DELETE等)。

通道(Channel)在AMQP是一個很重要的概念，大多數操作都是在通道這個層面展開的

我們完全可以用Connection就能完成通道的工作，為什麼還要引入通道？

試想:一個程式中有很多個執行緒需要從RabbitMQ中消費訊息，或者生產訊息，那麼必然需要建立很多個Connection，也就是多個TCP連線。

建立和銷燬TCP連線開銷很昂貴。 ofollow,noindex"> 所以RabbitMQ採用類似IO/">NIO的做法，選擇TCP連線複用。不僅可以減少效能開銷，同時也便於管理。

釋出訂閱模式

廣播模式 topic

所謂廣播指的是一條訊息將被所有的消費者進行處理。

直連模式 director

直連模式的特點主要就是routingkey的使用，如果現在該訊息就要求指定一個具備有指定Routingkey的操作者進行處理，那麼只需要兩個的Routingkey匹配即可。

可以將Routingkey比喻一個唯一標記，這樣就可以將訊息準確的推送到消費者手中了。

主題模式 fanout

主題模式類似於廣播模式與直連模式的整合操作，所有的消費者都可以接收到主題資訊，但是如果要想進行正確的處理，則一定需要有一個合適的Routingkey完成操作。

交換器相當於投遞包裹的郵箱，Routingkey相當於包裹的地址，BindingKey相當於包裹的目的地。

當填寫在包裹上的地址和要投遞的地址相匹配時，那麼這個包裹就會正確投遞到目的地，最後這個目的地的主人(佇列)可以保留這個包裹。

如果填寫地址出錯，郵遞員不能正確的投遞到目的地，包裹可能被退回給寄件人，也有可能被丟棄。

RabbitMQ官方文件和API都把Routingkey和BingdingKey都看做Routingkey下面程式碼中紅色部分就都當Routingkey使用

訊息生產者

public class MessageProducer {
private static final String EXCHANGE_NAME ="com.sunkun.topic";//訊息佇列名稱
private static final String HOST="192.168.1.105";
private static final int PORT=5672;
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();//建立一個連線工廠
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername("sunkun");
factory.setPassword("123456");
//factory.setVirtualHost(virtualHost) 使用虛擬主機的最大好處 可以區分不同使用者的操作空間每一個虛擬主機有一個自己的空間管理
Connection conn = factory.newConnection();//定義一個新的RabbitMQ的連線
Channel channel = conn.createChannel();//建立一個通訊的通道
//定義該通道要使用的佇列名稱 此時佇列已經建立過了
//第一個引數 佇列名稱（這個佇列可能存在也可能不存在）
//第二個引數 是否持久儲存
//第三個引數 此佇列是否為專用的佇列資訊
//第四個引數 是否允許自動刪除
//channel.queueDeclare(QUENE_NAME, true, false, true,null);
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
long start = System.currentTimeMillis();
System.out.println("訊息開始"+start);
for(int i=0;i<1000;i++){
String message = "sk - "+i;
if(i%2==0){
//MessageProperties.PERSISTENT_TEXT_PLAIN 訊息持久化
channel.basicPublish(EXCHANGE_NAME, "sk1", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());//進行訊息傳送
}else{
channel.basicPublish(EXCHANGE_NAME, "sk2", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());//進行訊息傳送
}

}
long end = System.currentTimeMillis();
System.out.println("訊息花費時間"+(end-start));
channel.close();
}
}

訊息消費者

public class MessageConsumer {
private static final String EXCHANGE_NAME ="com.sunkun.topic";//訊息佇列名稱
private static final String HOST="192.168.1.105";
private static final int PORT=15672;
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();//建立一個連線工廠
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername("sunkun");
factory.setPassword("123456");
Connection conn = factory.newConnection();//定義一個新的RabbitMQ的連線
Channel channel = conn.createChannel();//建立一個通訊的通道
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName = channel.queueDeclare().getQueue();//通過通道獲取一個佇列名稱
channel.queueBind(queueName, EXCHANGE_NAME, "sk2");//進行繫結處理
//在RabbitMQ裡面，所有的消費者資訊是通過一個回撥方法完成的
Consumer consumer = new DefaultConsumer(channel){//需要複寫指定的方法實現訊息處理
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
BasicProperties properties, byte[] body) throws IOException {
String message = new String(body);
System.out.println("消費者sk2:"+message);//可以啟動多個消費者
super.handleDelivery(consumerTag, envelope, properties, body);
}
};
channel.basicConsume(queueName,consumer);
}
}

RabbitMQ如何保證訊息的可靠性

1)持久化

持久化可以提高RabbitMQ的可靠性，防止在異常情況(重啟，關閉，宕機)下的資料丟失。

持久化可分為三個部分：交換器的持久化，佇列的持久化和訊息的持久化。

交換器的持久化：是通過宣告交換器時將druable引數設定為true來實現的。如果交換器不設定持久化，那麼在RabbitMQ重啟之後相關的交換器元資料會丟失，不過訊息不會丟失，只是不能將訊息傳送到這個交換器中了。

對於一個長期使用的交換器來說，建議其置為持久化。 (訊息不直接往佇列發，往exchange傳送可以實現廣播模式)

佇列的持久化：是通過宣告佇列時將durable引數置為true實現的。如果佇列不設定持久化，那麼在RabbitMQ服務重啟之後，相關佇列的元資料會丟失，此時資料也會丟失。

訊息的持久化：因為佇列的持久化能保證其本身的元資料不會因為異常情況而丟失，但是不能保證內部儲存的訊息不會丟失。要確保訊息不會丟失，需求將其設定為持久化。

通過將訊息的投遞模式( MessageProperties.PERSISTENT_TEXT_PLAIN )即可實現訊息的持久化

2)叢集

RabbitMQ的叢集本身不帶有所謂的HA機制以及負載均衡機制

本文主要講映象佇列

在持久化的訊息正確存入到RabbitMQ之後還需要一段時間(雖然時間很短，但不可忽視)才能存入磁碟中，如果這段時間發生了宕機，訊息儲存還沒來得及落盤，那麼這些訊息將會丟失。

這裡可以引入RabbitMQ的映象佇列機制，相當於配置了副本，如果主節點(master)在此特殊時間內掛掉，可以自動切換到從節點(Slave)，在實際生產環境中的關鍵業務佇列都會設定映象佇列。

提醒： 所謂的映象佇列只是進行資料的副本而已 ，在所謂的RabbitMQ叢集裡面並不支援HA機制以及所謂的負載均衡，如果說現在一臺主機掛掉了，那麼其他主機肯定無法進行合理讀取的。

如果想要安全的使用RabbitMQ就要繼續追加負載均衡元件，列如HAProxy LVS等等，如果要保證負載均衡元件的高可用，還應該繼續追加KeepAlive元件(就像tomcat實現負載均衡需要nginx一樣)。

3)生產者確認

除上面兩個問題外我們還遇到一個新問題：當訊息的生產者將訊息傳送出去之後，訊息到底有沒有正確的到達伺服器呢？

如果訊息到達伺服器之前就丟失，那麼持久化也解決不了問題，因為訊息就沒有到達伺服器，何談持久化呢。

通常會有兩種方法解決此問題一時事物機制，只有訊息被成功接收，事物才能提交成功，否則便可在捕獲異常之後進行事物回滾，於此同時可以進行訊息重發。

但使用事物機制會大大降低RabbitMQ的效能，我們一般採取傳送方確認機制。

傳送方確認機制：生產者將通道設定成confirm模式，一旦通道進入confirm模式，所有在該通道上面釋出的訊息都會被指派一個唯一的ID(從1開始)，

一旦訊息被投遞到所有的匹配佇列之後，RabbitMQ就會發送一個確認給生產者(包含訊息的唯一ID)，這就使得生產者知曉訊息已經到達目的地了。

如果訊息和佇列是持久化的，那麼訊息確認會在訊息寫入磁碟後發出。