RabbitMQ系列之---初識RabbitMQ
阿新 • • 發佈：2019-05-09
為什麼要使用RabbitMQ？
訊息佇列的作用
- 非同步呼叫
- 系統解耦
- 削峰限流
- 訊息通訊
訊息佇列的缺點
- 系統可用性降低
- 系統穩定性降低
- 分散式一致性問題（可靠訊息最終一致性的分散式事務方案解決）
RabbitMQ的優勢
- 支援高併發、高吞吐、效能好
- 有完善的後臺管理介面
- 它還支援叢集化、高可用部署架構、訊息高可靠支援
- RabbitMQ的開源社群很活躍，較高頻率的迭代版本，來修復發現的bug以及進行各種優化
- 最重要的是它是開源免費的。
RabbitMQ的缺點
- 它是基於erlang語言開發的，所以導致較為難以分析裡面的原始碼，也較難進行深層次的原始碼定製和改造，畢竟需要較為紮實的erlang語言功底才可以。
核心概念
- Server：又稱Broker，接受客戶端的連線，實現AMQP實體服務；
- Connection：連線，應用程式與Broker的網路連線；
- Channel：網路通道，也稱通道，幾乎所有的通道都在Channel中進行的，Channel是進行訊息讀寫的通道。客戶端可建立多個Channel，每個Channel代表一個會話任務；是建立在“真實的”TCP連線內的虛擬連線。AMQP命令都是通過通道傳送出去的。那麼我們為什麼需要通道呢？為什麼不直接通過TCP連線傳送AMQP命令呢？因為作業系統建立和銷燬TCP會話是非常昂貴的開銷，而且作業系統只能建立數量不多的TCP連線，很快就達到效能瓶頸，無法滿足高效能的需求。
- Message：訊息，伺服器和應用程式之間傳送的資料，由Properties和Body組成。Properties可以對訊息進行修飾，比如訊息的優先順序、等高階特性；Body則就是訊息體的內容；
- Virtual host：虛擬地址，用於進行邏輯隔離，最上層的訊息路由，並非物理概念。一個Virtual Host裡面有若干個Exchange和Queue，同一個Virtual Host裡面不能有相同名稱的Exchange或者Queue；
- Exchange：交換機，接收訊息，根據路由鍵轉發訊息到繫結的佇列；
交換機的型別：direct：直連，fanout：廣播，headers：以...開頭，topic：主題參考：Spring Boot RabbitMQ 四種交換器 - Binding：Exchange和Queue之間的虛擬連線，binding中可以包含routing key；
- Routing Key：一個路由規則，虛擬機器可用它來確定如何路由一個特定的訊息；
- Queue：也稱為Message Queue，訊息佇列，儲存訊息並將它們轉發給消費者。
幾種交換機
- Direct：直連交換機
所有傳送到Direct Exchange的訊息被轉發到RouteKey中指定的Queue
注意：Direct模式可以直接使用RabbitMQ自帶的Exchange：default，並以佇列名作為路由鍵。 Exchange，所以不需要將Exchange進行任何繫結(binding)操作，訊息傳遞時，RouteKey必須完全匹配才會被佇列接收，否則該訊息會被拋棄。
public class Producer4DirectExchange {
public static void main(String[] args) throws IOException, TimeoutException {
...
Channel channel = connection.createChannel();
String exchangeName = "test_direct_exchange";
String routingKey = "test.direct";
String msg = "Hello World RabbitMQ 4 Direct Exchange Message...";
channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
...
}
}public class Consumer4DirectExchange {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
...
Channel channel = connection.createChannel();
String exchangeName = "test_direct_exchange";
String exchangeType = "direct";
String queueName = "test_direct_queue";
String routingKey = "test.direct";
// 宣告交換機
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
// 宣告佇列
channel.queueDeclare(queueName, false, false, false, null);
// 交換機與佇列繫結
channel.queueBind(queueName, exchangeName, routingKey);
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
channel.basicConsume(queueName,true,queueingConsumer);
while (true) {
QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到訊息： " + msg);
}
}
}- Topic：主題交換機
所有傳送到Topic Exchange的訊息被轉發到所有關心RouteKey中指定的Queue上，Exchange將RouteKey和某個Topic進行模糊匹配，此時佇列需要繫結一個Topic。
注意：可以使用萬用字元進行模糊匹配
"#" 匹配一個或多個詞
"*" 匹配一個詞
"." 是單詞的分隔符
例如：
"log.#"：能夠匹配到"log.info.oa
"log.*"：只會匹配到"log.error"public class Producer4TopicExchange {
public static void main(String[] args) throws IOException, TimeoutException {
...
Channel channel = connection.createChannel();
String exchangeName = "test_topic_exchange";
String routingKey1 = "user.save";
String routingKey2 = "user.update";
String routingKey3 = "user.delete.ok";
String msg = "Hello World RabbitMQ 4 Topic Exchange Message...";
channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());
channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());
...
}
}public class Consumer4TopicExchange {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
...
Channel channel = connection.createChannel();
String exchangeName = "test_topic_exchange";
String exchangeType = "topic";
String queueName = "test_topic_queue";
String routingKey = "user.*";
channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, queueingConsumer);
while (true) {
QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到訊息： " + msg);
}
}
}- Fanout Exchange：廣播交換機
不處理路由鍵，只需要簡單的將佇列繫結到交換機；
傳送到交換機的訊息都會被轉發到與該交換機繫結的所有佇列上；
Fanout交換機轉發訊息是最快的。
public class Producer4FanoutExchange {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
...
Channel channel = connection.createChannel();
String exchangeName = "test_fanout_exchange";
String routingKey = "";
String msg = "Hello World RabbitMQ 4 Fanout Exchange Message...";
channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
...
}
}public class Consumer4FanoutExchange {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
...
Channel channel = connection.createChannel();
String exchangeName = "test_fanout_exchange";
String exchangeType = "fanout";
String queueName = "test_fanout_queue";
String routingKey = "";
channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, queueingConsumer);
while (true) {
QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到訊息： " + msg);
}
}
}參考資料：
- 石杉大神訊息中介軟體系列文章
- 慕課網《RabbitMQ訊息中介軟體技術精講》
- 書《RabbitMQ實戰-高效部署分散式訊息佇列》