1、MQ 的相關(guān)概念
1）什么是 MQ
MQ (message queue)，從字面意思上看，本質(zhì)是個隊列，F(xiàn)IFO 先入先出，只不過隊列中存放的內(nèi)容是 message 而已，還是一種跨進程的通信機制，用于上下游傳遞消息。在互聯(lián)網(wǎng)架構(gòu)中，MQ 是一種非常常見的上下游 “邏輯解耦 + 物理解耦” 的消息通信服務(wù)。使用了 MQ 之后，消息發(fā)送上游只需要依賴 MQ，不 用依賴其他服務(wù)
2）為什么要用 MQ
流量消峰
舉個例子，如果訂單系統(tǒng)最多能處理一萬次訂單，這個處理能力應(yīng)付正常時段的下單時綽綽有余，正常時段我們下單一秒后就能返回結(jié)果。但是在高峰期，如果有兩萬次下單操作系統(tǒng)是處理不了的，只能限制訂單超過一萬后不允許用戶下單。使用消息隊列做緩沖，我們可以取消這個限制，把一秒內(nèi)下的訂單分散成一段時間來處理，這時有些用戶可能在下單十幾秒后才能收到下單成功的操作，但是比不能下單的體驗要好。
應(yīng)用解耦
以電商應(yīng)用為例，應(yīng)用中有訂單系統(tǒng)、庫存系統(tǒng)、物流系統(tǒng)、支付系統(tǒng)。用戶創(chuàng)建訂單后，如果耦合調(diào)用庫存系統(tǒng)、物流系統(tǒng)、支付系統(tǒng)，任何一個子系統(tǒng)出了故障，都會造成下單操作異常。當(dāng)轉(zhuǎn)變成基于消息隊列的方式后，系統(tǒng)間調(diào)用的問題會減少很多，比如物流系統(tǒng)因為發(fā)生故障，需要幾分鐘來修復(fù)。在 這幾分鐘的時間里，物流系統(tǒng)要處理的內(nèi)存被緩存在消息隊列中，用戶的下單操作可以正常完成。當(dāng)物流系統(tǒng)恢復(fù)后，繼續(xù)處理訂單信息即可，中單用戶感受不到物流系統(tǒng)的故障，提升系統(tǒng)的可用性
異步處理
有些服務(wù)間調(diào)用是異步的，例如 A 調(diào)用 B，B 需要花費很長時間執(zhí)行，但是 A 需要知道 B 什么時候可以執(zhí)行完。
以前一般有兩種方式，A 過一段時間去調(diào)用 B 的查詢 api 查詢?；蛘?A 提供一個 callback api，B 執(zhí)行完之后調(diào)用 api 通知 A 服務(wù)。這兩種方式都不是很優(yōu)雅。
使用消息總線，可以很方便解決這個問題， A 調(diào)用 B 服務(wù)后，只需要監(jiān)聽 B 處理完成的消息，當(dāng) B 處理完成后，會發(fā)送一條消息給 MQ，MQ 會將此消息轉(zhuǎn)發(fā)給 A 服務(wù)。這樣 A 服務(wù)既不用循環(huán)調(diào)用 B 的查詢 api，也不用提供 callback api。同樣 B 服務(wù)也不用做這些操作，A 服務(wù)還能及時的得到異步處理成功的消息。
3）MQ 的分類
ActiveMQ
優(yōu)點：單機吞吐量萬級，時效性 ms 級，可用性高，基于主從架構(gòu)實現(xiàn)高可用性，較低的概率丟失數(shù)據(jù)。
缺點：官方社區(qū)現(xiàn)在對 ActiveMQ 5.x 維護越來越少，高吞吐量場景較少使用。
Kafka
大數(shù)據(jù)的殺手锏，談到大數(shù)據(jù)領(lǐng)域內(nèi)的消息傳輸，則繞不開 Kafka，這款為大數(shù)據(jù)而生的消息中間件，以其百萬級 TPS 的吞吐量名聲大噪，迅速成為大數(shù)據(jù)領(lǐng)域的寵兒，在數(shù)據(jù)采集、傳輸、存儲的過程中發(fā)揮著舉足輕重的作用。目前已經(jīng)被 LinkedIn，Uber, Twitter, Netflix 等大公司所采納。
優(yōu)點：性能卓越，單機寫入 TPS 約在百萬條 / 秒，最大的優(yōu)點，就是吞吐量高。時效性 ms 級，可用性非常高，kafka 是分布式的，一個數(shù)據(jù)多個副本，少數(shù)機器宕機，不會丟失數(shù)據(jù)，不會導(dǎo)致不可用，消費者采用 Pull 方式獲取消息，消息有序，通過控制能夠保證所有消息被消費且僅被消費一次；有優(yōu)秀的第三方 Kafka Web 管理界面 Kafka-Manager；在日志領(lǐng)域比較成熟，被多家公司和多個開源項目使用；功能支持：功能較為簡單，主要支持簡單的 MQ 功能，在大數(shù)據(jù)領(lǐng)域的實時計算以及日志采集被大規(guī)模使用。
缺點：Kafka 單機超過 64 個隊列 / 分區(qū)，Load 會發(fā)生明顯的飆高現(xiàn)象，隊列越多，load 越高，發(fā)送消息響應(yīng)時間變長，使用短輪詢方式，實時性取決于輪詢間隔時間，消費失敗不支持重試；支持消息順序，但是一臺代理宕機后，就會產(chǎn)生消息亂序，社區(qū)更新較慢。
RocketMQ
RocketMQ 出自阿里巴巴的開源產(chǎn)品，用 Java 語言實現(xiàn)，在設(shè)計時參考了 Kafka，并做出了自己的一 些改進。被阿里巴巴廣泛應(yīng)用在訂單，交易，充值，流計算，消息推送，日志流式處理，binglog 分發(fā)等場景。
優(yōu)點：單機吞吐量十萬級，可用性非常高，分布式架構(gòu)，消息可以做到 0 丟失，MQ 功能較為完善，還是分布式的，擴展性好，支持 10 億級別的消息堆積，不會因為堆積導(dǎo)致性能下降。
缺點：支持的客戶端語言不多，目前是 java 及 c++，其中 c++ 不成熟；社區(qū)活躍度一般，沒有在 MQ 核心中去實現(xiàn) JMS 等接口，有些系統(tǒng)要遷移需要修改大量代碼。
RabbitMQ
2007 年發(fā)布，是一個在 AMQP (高級消息隊列協(xié)議) 基礎(chǔ)上完成的，可復(fù)用的企業(yè)消息系統(tǒng)，是當(dāng)前最主流的消息中間件之一。
優(yōu)點：由于 erlang 語言的高并發(fā)特性，性能較好；吞吐量到萬級，MQ 功能比較完備，健壯、穩(wěn)定、易用、跨平臺、支持多種語言。如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，AJAX 文檔齊全；開源提供的管理界面非常棒，用起來很好用，社區(qū)活躍度高；更新頻率相當(dāng)高。
**缺點：**商業(yè)版需要收費，學(xué)習(xí)成本較高。
4）MQ 的選擇
Kafka
Kafka 主要特點是基于 Pull 的模式來處理消息消費，追求高吞吐量，一開始的目的就是用于日志收集和傳輸，適合產(chǎn)生大量數(shù)據(jù)的互聯(lián)網(wǎng)服務(wù)的數(shù)據(jù)收集業(yè)務(wù)。大型公司建議可以選用，如果有日志采集功能，肯定是首選 kafka 了。
RocketMQ
天生為金融互聯(lián)網(wǎng)領(lǐng)域而生，對于可靠性要求很高的場景，尤其是電商里面的訂單扣款，以及業(yè)務(wù)削峰，在大量交易涌入時，后端可能無法及時處理的情況。RoketMQ 在穩(wěn)定性上可能更值得信賴，這些業(yè)務(wù)場景在阿里雙 11 已經(jīng)經(jīng)歷了多次考驗，如果你的業(yè)務(wù)有上述并發(fā)場景，建議可以選擇 RocketMQ。
RabbitMQ
結(jié)合 erlang 語言本身的并發(fā)優(yōu)勢，性能好時效性微秒級，社區(qū)活躍度也比較高，管理界面用起來十分方便，如果你的數(shù)據(jù)量沒有那么大，中小型公司優(yōu)先選擇功能比較完備的 RabbitMQ。
2、RabbitMQ
1）RabbitMQ 的概念
RabbitMQ 是一個消息中間件，它接受并轉(zhuǎn)發(fā)消息。你可以把它當(dāng)做一個快遞站點，當(dāng)你要發(fā)送一個包裹時，你把你的包裹放到快遞站，快遞員最終會把你的快遞送到收件人那里，按照這種邏輯 RabbitMQ 是 一個快遞站，一個快遞員幫你傳遞快件。
RabbitMQ 與快遞站的主要區(qū)別在于，它不處理快件而是接收，存儲和轉(zhuǎn)發(fā)消息數(shù)據(jù)。
2）四大核心概念
生產(chǎn)者：產(chǎn)生數(shù)據(jù)發(fā)送消息的程序。
交換機：是 RabbitMQ 非常重要的一個部件，一方面它接收來自生產(chǎn)者的消息，另一方面它將消息推送到隊列中。交換機必須確切知道如何處理它接收到的消息，是將這些消息推送到特定隊列還是推送到多個隊列，亦或者是把消息丟棄，這個得有交換機類型決定。
隊列：隊列是 RabbitMQ 內(nèi)部使用的一種數(shù)據(jù)結(jié)構(gòu)，盡管消息流經(jīng) RabbitMQ 和應(yīng)用程序，但它們只能存儲在隊列中。隊列僅受主機的內(nèi)存和磁盤限制的約束，本質(zhì)上是一個大的消息緩沖區(qū)。許多生產(chǎn)者可以將消息發(fā)送到一個隊列，許多消費者可以嘗試從一個隊列接收數(shù)據(jù)。
消費者：大多時候是一個等待接收消息的程序。請注意生產(chǎn)者，消費者和消息中間件很多時候并不在同一機器上。同一個應(yīng)用程序既可以是生產(chǎn)者又是可以是消費者。
3）各個名詞介紹
Broker：接收和分發(fā)消息的應(yīng)用，RabbitMQ Server 就是 Message Broker。
Virtual host：出于多用戶和安全因素設(shè)計的，把 AMQP 的基本組件劃分到一個虛擬的分組中，類似于網(wǎng)絡(luò)中的 namespace 概念。當(dāng)多個不同的用戶使用同一個 RabbitMQ server 提供的服務(wù)時，可以劃分出多個 vhost，每個用戶在自己的 vhost 創(chuàng)建 exchange／queue 等。
Connection：publisher／consumer 和 broker 之間的 TCP 連接。
Channel：如果每一次訪問 RabbitMQ 都建立一個 Connection，在消息量大的時候建立 TCP Connection 的開銷將是巨大的，效率也較低。Channel 是在 connection 內(nèi)部建立的邏輯連接，如果應(yīng)用程序支持多線程，通常每個 thread 創(chuàng)建單獨的 channel 進行通訊，AMQP method 包含了 channel id 幫助客戶端和 message broker 識別 channel，所以 channel 之間是完全隔離的。Channel 作為輕量級的 Connection 極大減少了操作系統(tǒng)建立 TCP connection 的開銷。
Exchange：message 到達 broker 的第一站，根據(jù)分發(fā)規(guī)則，匹配查詢表中的 routing key，分發(fā)消息到 queue 中去。常用的類型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)。
Queue：消息最終被送到這里等待 consumer 取走。
Binding：exchange 和 queue 之間的虛擬連接，binding 中可以包含 routing key，Binding 信息被保存到 exchange 中的查詢表中，用于 message 的分發(fā)依據(jù)。
3、Hello world
在此測試之前，大家得先把RabbitMQ部署到三臺服務(wù)器/或者三臺虛擬機上，筆者是用的是三臺虛擬機
連接的時候，需要開啟 5672 端口。
依賴
pom.xml：
<!--指定 jdk 編譯版本-->
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>8</source>
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<!--rabbitmq 依賴客戶端-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.8.0</version>
</dependency>
<!--操作文件流的一個依賴-->
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.6</version>
</dependency>
</dependencies>
1）消息生產(chǎn)者
測試效果：
打開web管理界面查看
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//創(chuàng)建一個連接工廠
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("42.192.149.23");
factory.setUsername("admin");
factory.setPassword("123456");
//channel 實現(xiàn)了自動 close 接口 自動關(guān)閉 不需要顯示關(guān)閉
//創(chuàng)建連接
Connection connection = factory.newConnection();
//獲取信道
Channel channel = connection.createChannel();
/**
* 生成一個隊列
* 1.隊列名稱
* 2.隊列里面的消息是否持久化 也就是是否用完就刪除
* 3.該隊列是否只供一個消費者進行消費 是否進行共享 true 可以多個消費者消費
* 4.是否自動刪除 最后一個消費者端開連接以后 該隊列是否自動刪除 true 自動刪除
* 5.其他參數(shù)
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "hello world";
/**
* 發(fā)送一個消息
* 1.發(fā)送到那個交換機
* 2.路由的 key 是哪個
* 3.其他的參數(shù)信息
* 4.發(fā)送消息的消息體
*/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("消息發(fā)送完畢");
}
}
2）消息消費者
獲取 “生產(chǎn)者” 發(fā)出的消息：
測試效果：
只需要消息體
消息都是發(fā)一次，就完了，所以需要重啟生產(chǎn)者，消費者再拿
文章來源：http://www.zghlxwxcb.cn/news/detail-831241.html
import com.rabbitmq.client.*;
public class Consumer {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("42.192.149.23");
factory.setUsername("admin");
factory.setPassword("123456");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
System.out.println("等待接收消息.........");
//推送的消息如何進行消費的接口回調(diào)
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody());
System.out.println(message);
};
//取消消費的一個回調(diào)接口 如在消費的時候隊列被刪除掉了
CancelCallback cancelCallback = (consumerTag) -> {
System.out.println("消息消費被中斷");
};
/**
* 消費者消費消息 - 接受消息
* 1.消費哪個隊列
* 2.消費成功之后是否要自動應(yīng)答 true 代表自動應(yīng)答 false 手動應(yīng)答
* 3.消費者未成功消費的回調(diào)
* 4.消息被取消時的回調(diào)
*/
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
消息隊列-RabbitMQ：MQ作用分類、RabbitMQ核心概念及消息生產(chǎn)消費調(diào)試 到此完結(jié)，筆者歸納、創(chuàng)作不易，大佬們給個3連再起飛吧文章來源地址http://www.zghlxwxcb.cn/news/detail-831241.html
到了這里，關(guān)于消息隊列-RabbitMQ：MQ作用分類、RabbitMQ核心概念及消息生產(chǎn)消費調(diào)試的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
