導(dǎo)入

一個技術(shù)的衍生必然是為了解決現(xiàn)實出現(xiàn)的問題，在講這個問題之前我們先了解一下傳統(tǒng)開發(fā)中關(guān)于服務(wù)調(diào)用出現(xiàn)的問題（痛點）有哪些？

我們?yōu)槭裁匆褂肕Q？

①、同步——超時

在多服務(wù)體系架構(gòu)中，必然存在著多個服務(wù)之間的調(diào)用關(guān)系，當(dāng)用戶提交了訂單，訂單服務(wù)會調(diào)用支付服務(wù)執(zhí)行用戶的金錢操作，執(zhí)行完畢之后緊接著調(diào)用商品服務(wù)對商家的商品信息（庫存、成交量、收入等）進行更新，執(zhí)行完畢之后又調(diào)用物流服務(wù)（對接發(fā)貨公司、收發(fā)地帶你）對用戶買的商品進行物流實時同步。每一次的服務(wù)調(diào)用都要等待另一個服務(wù)的執(zhí)行完畢，整個流程下來很耗時。對于系統(tǒng)來說要求實時性要強，可立即得到結(jié)果，而同步調(diào)用還存在著其他問題：

資源浪費：調(diào)用鏈中每個服務(wù)在等待響應(yīng)過程中，不能釋放請求占用的資源，高并發(fā)場景下會極度浪費系統(tǒng)資源

級聯(lián)失?。喝绻?wù)提供者出現(xiàn)問題，所有調(diào)用方都會跟著出問題，如同多米諾骨牌一樣，迅速導(dǎo)致真?zhèn)€微服務(wù)群鼓掌

②、服務(wù)耦合高

訂單服務(wù)需要分別調(diào)用支付服務(wù)、商品服務(wù)、物流服務(wù)，調(diào)用者需要等待服務(wù)提供者響應(yīng)，但是如果作為上游服務(wù)的物流服務(wù)突然宕機，這樣會導(dǎo)致訂單服務(wù)也會出現(xiàn)問題，用戶下單失敗；并且如果每次加入新需求，此時如果還需要調(diào)用通信服務(wù)實時給用戶同步下單情況這樣一個需求，我們就要修改原來的代碼，耦合度很高

③、流量高峰

一般在秒殺或團搶活動中使用廣泛。比方說訂單系統(tǒng)經(jīng)過測試組接口測試發(fā)現(xiàn)最多只能承受100萬次請求，而在面對618、雙一這種購物狂潮的高峰期，如果一下來了500萬的請求系統(tǒng)時是無法處理的，可能會導(dǎo)致數(shù)據(jù)庫無法承受這么大的壓力，響應(yīng)變慢或者直接掛掉

有沒有什么辦法能夠幫助解決上面出現(xiàn)的問題呢？面對上述的種種問題，偉大的人類就提出了一個奇思妙想——把數(shù)據(jù)暫存。可不可把所有要傳輸?shù)南⒎旁谝粋€容器中，當(dāng)大量請求來的時候先把一部分的消息（邏輯）暫存在容器中，然后慢慢去處理？ 于是出現(xiàn)了消息隊列的概念

MQ簡介

MQ是什么？

Message Queue，消息隊列，是基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)中“先進先出”的一種數(shù)據(jù)結(jié)構(gòu)。把要傳輸?shù)南ⅲ〝?shù)據(jù)）放在隊列中，用隊列機制來實現(xiàn)消息傳遞——生產(chǎn)者生產(chǎn)消息把消息放入隊列，然后消費者去處理。消費者可以到指定隊列拉取消息，或者訂閱響應(yīng)的隊列，由MQ服務(wù)端給其推送消息。

為什么要使用隊列這種數(shù)據(jù)結(jié)構(gòu)呢？

我們知道隊列具有先進先出的特點。而消息隊列就是將消息放到隊列里，用隊列做存儲消息的介質(zhì)，看作是一個容器。那這里的消息我們可以指代文本字符串，也可以是更復(fù)雜的嵌入對象。消息的發(fā)送放稱為生產(chǎn)者，消息的接收方稱為消費者

MQ要解決什么問題？

結(jié)合前面講到應(yīng)用場景出現(xiàn)的問題，我們來針對性討論一下方案：

①、同步請求—>異步請求

通過引入MQ之后，我為了提升系統(tǒng)響應(yīng)性能，我們可以它改造為異步，那異步請求有什么好處？它是如何解決同步出現(xiàn)的問題？

異步調(diào)用的本質(zhì)是一種事件驅(qū)動模式：

• 耦合度低（不需要調(diào)用對方，只需要發(fā)布事件，事件去響應(yīng)即可）
• 吞吐量提升（不需要等待對方，執(zhí)行時間更短，吞吐量更大）
• 故障隔離（如果出現(xiàn)故障更容易排查）
• 流量削峰（broker做緩存）

②、高耦合—>低耦合

如果使用消息隊列，當(dāng)訂單服務(wù)執(zhí)行完成之后，發(fā)送一條消息到隊列中，其余三個服務(wù)讀取到這條消息，那么它立刻開始進行業(yè)務(wù)的執(zhí)行。使用了消息隊列后，消息的發(fā)送方和接收方并不需要知道彼此，這樣相互之間也就是沒有直接關(guān)系，即解耦。

③、削峰

這種峰值流量場景一般是集中于一小段時間內(nèi)，為了防止系統(tǒng)在這個峰值時間內(nèi)被流量沖垮，可以采用消息隊列來削弱峰值流量，此時的消息隊列就可以理解為是一個緩沖區(qū)，比方說系統(tǒng)只能處理100萬請求，但此時同時有500萬請求來臨，我們就可以把把多余的400萬請求先放到隊列中，等系統(tǒng)根據(jù)自己處理請求的能力去消息隊列去消費。

一般用來解決應(yīng)用解耦，異步消息，流量削峰等問題，目的是為了實現(xiàn)高性能、高可用、可伸縮和最終一致性架構(gòu)

MQ的產(chǎn)品有哪些？

RabbitMQ、ActiveMQ、RocketMQ、ZeroMQ、Kafka、IBM WebSphere 等

通過我提出來的應(yīng)用場景想必大家對于為什么要使用MQ有了初步的了解，接下來我們重點討論一下RabbitMQ

RabbitMQ

RabbitMQ是什么？

RabbitMQ是一個由Erlang語言開發(fā)的AMQP的開源中間件

補充：AMQP是什么？

AMQP我們可以看作是一種協(xié)議活規(guī)范，而RabbitMQ是基于這個協(xié)議下的實現(xiàn)框架。類似于：JDBC和mysql

RabbitMQ的工作原理是什么？

架構(gòu)圖

整個消息傳輸流程為：

1. 生產(chǎn)者將消息發(fā)送到交換機；

2. 交換機根據(jù)路由鍵將消息路由到相應(yīng)的隊列；

3. 消費者從隊列中獲取消息進行處理。

RabbitMQ有哪些工作模式？

官方網(wǎng)站：RabbitMQ Tutorials — RabbitMQ

①、Simple（簡單工作模式）

特點：一個隊列只有一個消費者

消息分發(fā)的方式。不同工作模式指的是消息路由的策略和方式不同

內(nèi)部使用的默認交換機

生產(chǎn)者將消息發(fā)送到隊列，消費者從隊列取出消息

②、Work Queues（工作模式）

特點：多個消費者監(jiān)聽同一個隊列

在一個隊列中如果有多個消費者，消費者之間對于同一個消息的關(guān)系是競爭關(guān)系，同一條消息只能由一個消費者消費。但是分擔(dān)壓力，比方說有10條消息，C1處理13579消息，C2處理246810消息。順序取消息，C1取一條，C2取一條，消費規(guī)則是輪詢

應(yīng)用場景：

③、Pub/Sub（訂閱模式）

特點：多個消息隊列，每個消息隊列有一個消費者監(jiān)聽

X：交換機，

生產(chǎn)者生產(chǎn)了消息發(fā)給交換機，交換機路由分發(fā)給不同的消費者，消費者監(jiān)聽獲取消息。一個消息可以被多個消費者同時接收消費

每個消費者都監(jiān)聽自己的隊列

使用場景：

A服務(wù)可以通過異常處理，如果A服務(wù)發(fā)送后出現(xiàn)問題可以回滾，什么都有可能出現(xiàn)問題，而我們的目的是保持一致性

④、Routing（路由模式）

特點：一個交換機綁定多個消息隊列，每個消息隊列都有自己唯一的key，每個消息隊列有一個消費者監(jiān)聽

Topics（通配符/主題模式）

*：一個單詞

#：零個或多個單詞

實戰(zhàn)演練

依賴

<dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>
</dependencies>

生產(chǎn)者

package com.itheima.producer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.創(chuàng)建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //2.設(shè)置參數(shù)
        factory.setHost("114.115.170.214");
        factory.setPort(5672);
        factory.setVirtualHost("/itcast");
        factory.setUsername("admin");
        factory.setPassword("admin");

        //3.創(chuàng)建連接 Connection
        Connection connection = factory.newConnection();

        //4.創(chuàng)建Channel
        Channel channel = connection.createChannel();

        //5.創(chuàng)建隊列Queue
        channel.queueDeclare("denglimei",true,false,false,null);

        //要發(fā)送的消息
        String body ="hello rabbitmq~~~";

        //6.發(fā)送消息
        channel.basicPublish("","denglimei",null,body.getBytes());
         //7.釋放資源
        channel.close();
        connection.close();
    }
}

消費者

package com.itheima.consumer;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.創(chuàng)建連接工廠
        ConnectionFactory factory = new ConnectionFactory();

        //2.設(shè)置參數(shù)
        factory.setHost("114.115.170.214");
        factory.setPort(5672);
        factory.setVirtualHost("/itcast");
        factory.setUsername("admin");
        factory.setPassword("admin");

        //3.創(chuàng)建連接 Connection
        Connection connection = factory.newConnection();

        //4.創(chuàng)建Channel
        Channel channel = connection.createChannel();

        //5.創(chuàng)建隊列Queue
        channel.queueDeclare("hello_world",true,false,false,null);

        //匿名內(nèi)部類
        Consumer consumer = new DefaultConsumer(channel){
            /*
             * 回調(diào)方法，當(dāng)收到消息后，會自動執(zhí)行該方法
             * 1.consumerTag：標(biāo)識
             * 2.envelope：獲取一些信息，交換機，路由key
             * 3.properties：配置信息
             * 4.body：數(shù)據(jù)
             * */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                long start = System.currentTimeMillis();
                for (int i = 0; i < 10000; i++) {
                    System.out.println("第"+i+"條");
                }
                long end = System.currentTimeMillis();

                System.out.println(start-end);
            }
        };

        //接收消息
        channel.basicConsume("hello_world",true,consumer);

        //注意：消費者作為監(jiān)聽者，不要去關(guān)閉資源，否則如何監(jiān)聽資源？
    }
}

總結(jié)

本次先對MQ和衍生出來的RabbitMQ先做了基本介紹，后續(xù)會針對RabbitMQ如何在項目中如何進行削峰、如何解決死信隊列等內(nèi)容做具體介紹，敬請期待~文章來源地址http://www.zghlxwxcb.cn/news/detail-594155.html

如果有想要交流的內(nèi)容歡迎在評論區(qū)進行留言，如果這篇文檔受到了您的喜歡那就留下你點贊+收藏+評論腳印支持一下博主~

到了這里，關(guān)于RabbitMQ到底為什么要使用它？的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！