本文轉載自【微信公眾號：java進階架構師，ID:java_jiagoushi】經微信公眾號授權轉載，如需轉載與原文作者聯繫
一、MQ用途
1、同步變異步消息
場景：用戶下單完成後，發送郵件和簡訊通知。
運用消息隊列之後，用戶下單完之後，下單信息寫入資料庫，再寫入消息隊列，發送郵件和發送簡訊各自去消息隊列進行讀取，節省時間，提高效率。
2、應用解耦
場景：用戶下單後，訂單系統需要多渠道通知用戶。
下單服務系統：用戶使用下單服務後，將下單信息寫入資料庫，下單成功。簡訊服務系統：用戶下單後，將簡訊信息寫入消息隊列，以發送簡訊信息通知用戶交易信息。郵件服務系統：用戶下單後，將郵件信息寫入消息隊列，以發送郵件信息通知用戶交易信息。這樣，如果微信通知不能正常使用，也不影響用戶下單，用戶下單後，只用把下單通知信息寫入消息隊列，不用關心後續操作，實現了訂單系統和通知系統的解耦。
3、流量削峰
一般在秒殺或者團購活動中使用。
場景：秒殺活動，一般會因為流量過大，導致流量暴增，應用掛掉。針對這個問題，一般需要在應用前端加入消息隊列。
可以控制活動的人數可以緩解短時間內高流量壓垮應用用戶的請求，伺服器接收後，首先寫入消息隊列，如果消息隊列的數量大於最大的數量，則直接拋棄用戶請求或者跳轉錯誤頁面。
二、RabbitMQ原理介紹
如圖所示：
各組件意義如下：
三、RabbitMQ應用
RabbitMQ包依賴（spring-boot-starter-amqp）：
<!-- rabbitMQ的依賴。rabbitmq已經被spring-boot做了整合訪問實現。spring cloud也對springboot做了整合邏輯。所以rabbitmq的依賴可以在spring cloud中直接使用。 --><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId></dependency>
1、Direct交換器
是一種點對點，實現發布/訂閱標準的交換器。Producer發送消息到RabbitMQ中，MQ中的Direct交換器接受到消息後，會根據Routing Key來決定這個消息要發送到哪一個隊列中。Consumer則負責註冊一個隊列監聽器，來監聽隊列的狀態，當隊列狀態發生變化時，消費消息。註冊隊列監聽需要提供交換器信息，隊列信息和路由鍵信息。
這種交換器通常用於點對點消息傳輸的業務模型中。如電子郵箱。
如下圖所示日誌處理MQ示例：
Producer全局配置文件：
spring.application.name=direct-producerserver.port=8082# 必要配置# 配置rabbitmq連結相關信息。key都是固定的。是springboot要求的。# rabbitmq安裝位置spring.rabbitmq.host=localhost# rabbitmq的埠spring.rabbitmq.port=5672# rabbitmq的用戶名spring.rabbitmq.username=test# rabbitmq的用戶密碼spring.rabbitmq.password=123456# 可選配置# 配置producer中操作的Queue和Exchange相關信息的。key是自定義的。為了避免硬編碼（代碼中可以寫死）。# exchange的命名。交換器名稱可以隨意定義。mq.config.exchange=log.direct# 路由鍵， 是定義某一個路由鍵。info級別日誌使用的queue的路由鍵。mq.config.queue.info.routing.key=log.info.routing.key# 路由鍵，error級別日誌使用的queue的路由鍵。mq.config.queue.error.routing.key=log.error.routing.key
Producer消息發送類：
/*** 消息發送者 - Producer。 * @Component Producer類型的對象，必須交由Spring容器管理。 * 使用SpringBoot提供的AMQP啟動器，來訪問rabbitmq的時候，都是通過AmqpTemplate來實現的。 * 如果全局配置文件中，配置了rabbitmq相關內容，且工程依賴了starter-amqp，則spring容器自動創建AmqpTemplate對象。 */@Componentpublic class Sender { @Autowired private AmqpTemplate rabbitAmqpTemplate; //exchange 交換器名稱 @Value("${mq.config.exchange}") private String exchange; //routingkey 路由鍵 @Value("${mq.config.queue.info.routing.key}") private String routingkey; /* * 發送消息的方法 */ public void send(LogMessage msg){ /** * convertAndSend - 轉換並發送消息的template方法。 * 是將傳入的普通java對象，轉換為rabbitmq中需要的message類型對象，並發送消息到rabbitmq中。 * 參數一：交換器名稱。 類型是String * 參數二：路由鍵。 類型是String * 參數三：消息，是要發送的消息內容對象。類型是Object */ this.rabbitAmqpTemplate.convertAndSend(this.exchange, this.routingkey, msg); }}
Producer實體類：
/*** 消息內容載體，在rabbitmq中，存儲的消息可以是任意的java類型的對象。 * 強制要求，作為消息數據載體的類型，必須是Serializable的。 * 如果消息數據載體類型未實現Serializable，在收發消息的時候，都會有異常發生。 */public class LogMessage implements Serializable { private Long id; private String msg; private String logLevel; private String serviceType; private Date createTime; private Long userId; public LogMessage() { super(); } public LogMessage(Long id, String msg, String logLevel, String serviceType, Date createTime, Long userId) { super(); this.id = id; this.msg = msg; this.logLevel = logLevel; this.serviceType = serviceType; this.createTime = createTime; this.userId = userId; } @Override public String toString() { return "LogMessage [id=" + id + ", msg=" + msg + ", logLevel=" + logLevel + ", serviceType=" + serviceType + ", createTime=" + createTime + ", userId=" + userId + "]"; } public Long getId() { return id; } public void setId(Long id) { this.id = id; } public String getMsg() { return msg; } public void setMsg(String msg) { this.msg = msg; } public String getLogLevel() { return logLevel; } public void setLogLevel(String logLevel) { this.logLevel = logLevel; } public String getServiceType() { return serviceType; } public void setServiceType(String serviceType) { this.serviceType = serviceType; } public Date getCreateTime() { return createTime; } public void setCreateTime(Date createTime) { this.createTime = createTime; } public Long getUserId() { return userId; } public void setUserId(Long userId) { this.userId = userId; }}
Producer消息產生測試類：
/*** Direct交換器 * Producer測試。 * 注意： * 在rabbitmq中，consumer都是listener監聽模式消費消息的。 * 一般來說，在開發的時候，都是先啟動consumer，確定有什麼exchange、queue、routing-key，然後再啟動producer。 * 然後再啟動producer發送消息，。 */@RunWith(SpringRunner.class)@SpringBootTest(classes=SpringbootServerApplication.class)public class QueueTest { @Autowired private Sender sender; /* * 測試消息隊列 */ @Test public void testSend()throws Exception{ Long id = 1L; while(true){ Thread.sleep(1000); this.sender.send(new LogMessage(id,"test log", "info", "訂單服務", new Date(), id)); id++; } }}
Consumer全局配置：
spring.application.name=direct-consumerserver.port=8083spring.rabbitmq.host=localhostspring.rabbitmq.port=5672spring.rabbitmq.username=testspring.rabbitmq.password=123456# 自定義配置。 配置交換器exchange、路由鍵routing-key、隊列名稱 queue name；在RabbitMQ中隊列的生成# 交換器名稱mq.config.exchange=log.direct# info級別queue的名稱mq.config.queue.info=log.info# info級別的路由鍵mq.config.queue.info.routing.key=log.info.routing.key# error級別queue的名稱mq.config.queue.error=log.error# error級別的路由鍵mq.config.queue.error.routing.key=log.error.routing.key
Consumer消費者：
/*** 消息接收者 - consumer * * @RabbitListener - 可以註解類和方法。 * 註解類，當表當前類的對象是一個rabbit listener。 * 監聽邏輯明確，可以由更好的方法定義規範。 * 必須配合@RabbitHandler才能實現rabbit消息消費能力，一個類可以有多個方法，但是僅有一個方法註解@RabbitHandler。 * 註解方法，代表當前方法是一個rabbit listener處理邏輯。 * 方便開發，一個類中可以定義若干個listener邏輯。 * 方法定義規範可能不合理。如：一個方法的處理邏輯太多，造成方法的bad smell。 * * @RabbitListener - 代表當前類型是一個rabbitmq的監聽器。 * bindings:綁定隊列 * @QueueBinding - @RabbitListener.bindings屬性的類型。綁定一個隊列。 * value:綁定隊列， Queue類型。 * exchange:配置交換器， Exchange類型。 * key:路由鍵，字符串類型。 * * @Queue - 隊列。 * value:隊列名稱 * autoDelete:是否是一個臨時隊列。 * true ：當所有的consumer關閉後，自動刪除queue。 * false：當任意一個consumer啟動並創建queue後，如果queue中有消息未消費，無論是否有consumer繼續執行，都保存queue。 * * @Exchange - 交換器 * value:為交換器起個名稱 * type:指定具體的交換器類型 */@Component@RabbitListener( bindings=@QueueBinding( value=@Queue(value="${mq.config.queue.error}",autoDelete="false"), exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.DIRECT), key="${mq.config.queue.error.routing.key}" ) )public class ErrorReceiver { /** * 消費消息的方法。採用消息隊列監聽機制 * @RabbitHandler - 代表當前方法是監聽隊列狀態的方法，就是隊列狀態發生變化後，執行的消費消息的方法。 * 方法參數。就是處理的消息的數據載體類型。 */ @RabbitHandler public void process(LogMessage msg){ System.out.println("Error..........receiver: "+msg); }}
2、Topic交換器
主題交換器，也稱為規則匹配交換器。是通過自定義的模糊匹配規則來決定消息存儲在哪些隊列中。當Producer發送消息到RabbitMQ中時，MQ中的交換器會根據路由鍵來決定消息應該發送到哪些隊列中。Java知音公眾號內回復「後端面試」，送你一份Java面試題寶典
Consumer同樣是註冊一個監聽器到隊列，監聽隊列狀態，當隊列狀態發生變化時，消費消息。註冊監聽器需要提供交換器信息，隊列信息和路由鍵信息。
如下圖所示日誌處理MQ示例：
Producer公共配置文件：
spring.application.name=topic-producerspring.rabbitmq.host=192.168.1.122spring.rabbitmq.port=5672spring.rabbitmq.username=testspring.rabbitmq.password=123456mq.config.exchange=log.topic
Producer的User實體日誌發送類：
/*** 消息發送者 */@Componentpublic class UserSender { @Autowired private AmqpTemplate rabbitAmqpTemplate; //exchange 交換器名稱 @Value("${mq.config.exchange}") private String exchange; /* * 發送消息的方法 */ public void send(String msg){ //向消息隊列發送消息 //參數一：交換器名稱。 //參數二：路由鍵 //參數三：消息 this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.debug", "user.log.debug....."+msg); this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.info", "user.log.info....."+msg); this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.warn","user.log.warn....."+msg); this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.error", "user.log.error....."+msg); }}
Producer的Order實體日誌發送類：
/*** 消息發送者 */@Componentpublic class OrderSender { @Autowired private AmqpTemplate rabbitAmqpTemplate; //exchange 交換器名稱 @Value("${mq.config.exchange}") private String exchange; /* * 發送消息的方法 */ public void send(String msg){ //向消息隊列發送消息 //參數一：交換器名稱。 //參數二：路由鍵 //參數三：消息 this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.debug", "order.log.debug....."+msg); this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.info", "order.log.info....."+msg); this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.warn","order.log.warn....."+msg); this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.error", "order.log.error....."+msg); }}
Producer測試類：
/*** 消息隊列測試類 * @author Administrator * */@RunWith(SpringRunner.class)@SpringBootTest(classes=SpringbootServerApplication.class)public class QueueTest { @Autowired private UserSender usersender; @Autowired private ProductSender productsender; @Autowired private OrderSender ordersender; /* * 測試消息隊列 */ @Test public void test() throws InterruptedException{ while(true){ Thread.sleep(1000); this.usersender.send("UserSender.....");this.ordersender.send("OrderSender......"); } }}
可以看出Producer的發送和Direct沒有區別，Consumer的全局配置文件：
spring.application.name=topic-consumerspring.rabbitmq.host=192.168.1.122spring.rabbitmq.port=5672spring.rabbitmq.username=testspring.rabbitmq.password=123456mq.config.exchange=log.topicmq.config.queue.info=log.infomq.config.queue.error=log.errormq.config.queue.logs=log.all
Consumer中的info日誌消費者：
@Component@RabbitListener(bindings=@QueueBinding( value=@Queue(value="${mq.config.queue.info}",autoDelete="true"), exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.TOPIC), key="*.log.info" ) )public class InfoReceiver { @RabbitHandler public void process(String msg){ System.out.println("......Info........receiver: "+msg); }}
Consumer中的全體日誌消費者：
/*** 和direct交換器的區別是：Exchange的類型為TOPIC。 * 全日誌處理。 */@Component@RabbitListener( bindings=@QueueBinding( value=@Queue(value="${mq.config.queue.logs}",autoDelete="true"), exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.TOPIC), key="*.log.*" ) )public class LogsReceiver { @RabbitHandler public void process(String msg){ System.out.println("......All........receiver: "+msg); }}
3、Fanout交換器
廣播交換器。這種交換器會將接收到的消息發送給綁定的所有隊列中。當Producer發送消息到RabbitMQ時，交換器會將消息發送到已綁定的所有隊列中，這個過程交換器不會嘗試匹配路由鍵，所以消息中不需要提供路由鍵信息。
Consumer仍舊註冊監聽器到隊列，監聽隊列狀態，當隊列狀態發生變化，消費消息。註冊監聽器需要提供交換器信息和隊列信息。擴展：RocketMQ匯總
如下圖所示簡訊、APP推送的MQ示例：
由於Producer的測試類和以上無差別，不再贅述，如下Producer的發送類：
/*** 消息發送者 * fanout交換器 - * 使用fanout交換器的時候，交換器是忽略routing-key的匹配。 * 因為廣播不需要考慮路由鍵的匹配，只考慮在Exchange上綁定了多少個queue，這個由Consumer的配置決定。 * 會將接受到的消息發送到所有的綁定的queue中，進行消息的緩存。 */@Componentpublic class Sender { @Autowired private AmqpTemplate rabbitAmqpTemplate; //exchange 交換器名稱 @Value("${mq.config.exchange}") private String exchange; /* * 發送消息的方法 */ public void send(String msg){ //向消息隊列發送消息 //參數一：交換器名稱。 //參數二：路由鍵 無需填寫，填寫了也無效 //參數三：消息 this.rabbitAmqpTemplate.convertAndSend(this.exchange,"A", msg); }}
如下所示Consumer的SMS消費類：
/*** 使用fanout交換器的時候，可以在consumer中省略routing-key的配置。 * 因為fanout交換器忽略routing-key的匹配，即使配置當type=ExchangeTypes.FANOUT時也無效。 */@Component@RabbitListener( bindings=@QueueBinding( value=@Queue(value="${mq.config.queue.sms}",autoDelete="true"), exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.FANOUT) ) )public class SmsReceiver { @RabbitHandler public void process(String msg){ System.out.println("Sms........receiver: "+msg); }}
如Consumer的Publish消費類：
@Component@RabbitListener(bindings=@QueueBinding( value=@Queue(value="${mq.config.queue.push}",autoDelete="true"), exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.FANOUT) ) )public class PushReceiver { @RabbitHandler public void process(String msg){ System.out.println("Push..........receiver: "+msg); }}
四、RabbitMQ消息可靠性處理
前面內容，如果consumer未啟動，而producer發送了消息。則消息會丟失。如果consumer先啟動，創建queue後，producer發送消息可以正常消費。那麼當所有的consumer宕機的時候，queue會auto-delete，消息仍舊會丟失。這種情況，消息不可靠。有丟失的可能。
Rabbitmq的消息可靠性處理，分為兩部分。
消息不丟失。當consumer全部宕機後，消息不能丟失。------持久化解決消息不會錯誤消費。當consumer獲取消息後，萬一consumer在消費消息的過程中發生了異常，如果rabbitmq一旦發送消息給consumer後立刻刪除消息，也會有消息丟失的可能。-------確認機制解決1、消息持久化
@Queue註解中的屬性 - autoDelete：當所有消費客戶端連接斷開後，是否自動刪除隊列 。true：刪除 false：不刪除@Exchange註解中的屬性 - autoDelete：當交換器所有的綁定隊列都不再使用時，是否自動刪除交換器（更粗粒度，不建議）。true：刪除 false：不刪除2、消息確認機制 ACK - acknowledge
什麼是消息確認機制？
如果在消息處理過程中，消費者的伺服器在處理消息時發生異常，那麼這條正在處理的消息就很可能沒有完成消息的消費，如果RabbitMQ在Consumer消費消息後立刻刪除消息，則可能造成數據丟失。為了保證數據的可靠性，RabbitMQ引入了消息確認機制。
消息確認機制是消費者Consumer從RabbitMQ中收到消息並處理完成後，反饋給RabbitMQ的，當RabbitMQ收到確認反饋後才會將此消息從隊列中刪除。如果某Consumer在處理消息時出現了網絡不穩定，伺服器異常等現象時，那麼就不會有消息確認反饋，RabbitMQ會認為這個消息沒有正常消費，會將消息重新放入隊列中。如果在Consumer集群環境下，RabbitMQ未接收到Consumer的確認消息時，會立即將這個消息推送給集群中的其他Consumer，保證不丟失消息。如果Consumer沒有確認反饋，RabbitMQ將永久保存消息。 消息確認機制默認都是開啟狀態的，同時不推薦關閉消息確認機制。注意：如果Consumer沒有處理消息確認，將導致嚴重後果。如：所有的Consumer都沒有正常反饋確認信息，並退出監聽狀態，消息則會永久保存，並處於鎖定狀態，直到消息被正常消費為止。消息的發送者Producer如果持續發送消息到RabbitMQ，那麼消息將會堆積，持續佔用RabbitMQ所在伺服器的內存，導致「內存洩漏」問題。
消息確認機制處理方案：
編碼異常處理（推薦）
通過編碼處理異常的方式，保證消息確認機制正常執行。這種處理方案也可以有效避免消息的重複消費。
異常處理，不是讓Consumer編碼catch異常後，直接丟棄消息，或反饋ACK確認消息。而是做異常處理的。該拋的異常，還得拋，保證ACK機制的正常執行。或者使用其他的手法，實現消息的再次處理。如：catch代碼塊中，將未處理成功的消息，重新發送給MQ。如：catch代碼中，本地邏輯的重試（使用定時線程池重複執行任務3次。）
配置重試次數處理
通常來說，消息重試3次以上未處理成功，就是Consumer開發出現了嚴重問題。需要修改Consumer代碼，提升版本/打補丁之類的處理方案。
通過全局配置文件，開啟消息消費重試機制，配置重試次數。當RabbitMQ未收到Consumer的確認反饋時，會根據配置來決定重試推送消息的次數，當重試次數使用完畢，無論是否收到確認反饋，RabbitMQ都會刪除消息，避免內存洩漏的可能。具體配置如下：
#開啟重試spring.rabbitmq.listener.retry.enabled=true#重試次數，默認為3次spring.rabbitmq.listener.retry.max-attempts=5
五、常用MQ產品對比和選擇
社區活躍度：RabbitMQ > ActiveMQ = RocketMQ > kafka
消息持久化：RabbitMQ、ActiveMQ、RocketMQ、kafka都支持持久化。ZeroMQ不支持持久化。
高並發：RabbitMQ = kafka > RocketMQ > ActiveMQ。RabbitMQ高並發是基於ErLang的。ErLang本身就是針對高並發提供的一種開發腳本語言。
吞吐量：RabbitMQ = kafka > RocketMQ > ActiveMQ。小型項目（並發吞吐低於萬級別）使用ActiveMQ。中型項目（並發吞吐10萬~100萬級），可選RocketMQ、ActiveMQ。大型項目優先考慮RabbitMQ和Kafka。
綜合技術：RabbitMQ和kafka最好。RocketMQ次之。ActiveMQ最弱。如：可靠性、路由、集群、事務、高可用隊列、消息可靠排序、持久化、可視化管理工具等。
RabbitMQ和Kafka選擇：建議Kafka針對日誌處理。其他使用RabbitMQ。商業項目中，如果現有的系統架構已經使用了某一個MQ產品，且沒有業務和性能上的問題，不推薦切換MQ產品。