高并發(fā)的核心技術(shù) - 消息中間件（MQ）

MQ簡介

• 什么是MQ
  跨進程的消息隊列，主要角色包括生產(chǎn)者與消費者。
  生產(chǎn)者只負責生產(chǎn)信息，無法感知消費者是誰，消息怎么處理，處理結(jié)果是什么。
  消費者負責接收及處理消息，無法感知生產(chǎn)者是誰，怎么產(chǎn)生的。

• Mq能做什么？
  MQ 特性一般有異步，吞吐量大 ，延時低；
  適合做：

  1. 投遞異步通知。
  2. 限流，削峰谷。
  3. 可靠事件，處理數(shù)據(jù)一致性。
  4. 利用一些特性，可以做定時任務(wù)。
     等….

由于MQ是異步處理消息的，所以MQ不適合做同步處理操作，如果需要及時的返回處理結(jié)果請不要用MQ；

• MQ 個系統(tǒng)帶來了什么?

  缺點：增加了系統(tǒng)的復(fù)雜性，除了代碼組件接入以外還需要考慮，高可用，集群，消息的可靠性等問題！

  生產(chǎn)者：消息發(fā)送怎么保證可靠性，怎么保證不重復(fù)！

  消費者：怎么保證冪等性，接收到重復(fù)消息怎么處理！

  還有會帶來的處理延時等問題!

優(yōu)點： 解耦，利用MQ我們可以很好的給我們系統(tǒng)解耦，特別是分布式/微服系統(tǒng)！
原來的同步操作，可以用異步處理，也可以帶來更快的響應(yīng)速度；

• 哪些場景可以使用MQ

場景 （1）
系統(tǒng)解耦，用戶系統(tǒng)或者其他系統(tǒng)需要發(fā)送短信可以通過 MQ 執(zhí)行；很好的將 用戶系統(tǒng) 和 短信系統(tǒng)進行解耦；

場景（2）

順序執(zhí)行的任務(wù)場景，假設(shè) A B C 三個任務(wù)，B需要等待 A完成才去執(zhí)行，C需要等待B完成才去執(zhí)行；

我見過一些同學(xué)的做法是 ，用 三個定時器 錯開時間去執(zhí)行的，假設(shè) A定時器 9 點執(zhí)行， B 定時器 10 點執(zhí)行 ， C 11 點執(zhí)行 ， 類似這樣子；

這樣做其實是 不安全的， 因為 后一個任務(wù) 無法知道 前一個任務(wù)是否 真的執(zhí)行了！ 假設(shè) A 宕機了， 到 10 點 B 定時去 執(zhí)行，這時候 數(shù)據(jù)就會產(chǎn)生異常！

當我們 引入 MQ 后 可以這么做， A執(zhí)行完了 發(fā)送 消息給 B ，B收到消息后 執(zhí)行，C 類似，收到 B消息后執(zhí)行；

場景（3）

支付網(wǎng)關(guān)的通知，我們的系統(tǒng)常常需要接入支付功能，微信或者支付寶通常會以回調(diào)的形式通知我們系統(tǒng)支付結(jié)果。

我們可以將我們的支付網(wǎng)關(guān)獨立出來，通過MQ通知我們業(yè)務(wù)系統(tǒng)進行處理，這樣處理有利于系統(tǒng)的解耦和擴展！

假設(shè)我們還有一個積分系統(tǒng)，用戶支付成功，給用戶添加積分。只需要積分系統(tǒng)監(jiān)聽這個消息，并處理積分就好，無需去修改再去修改網(wǎng)關(guān)層代碼！

如果沒有使用MQ ，我是不是還得去修改網(wǎng)關(guān)系統(tǒng)的代碼，遠程調(diào)用增加積分的接口？

這就是使用了MQ的好處，解耦和擴展！

當然我們的轉(zhuǎn)發(fā)規(guī)則也要保證每個感興趣的隊列能獲取到消息！

場景（4）

微服/分布式系統(tǒng)，分布式事務(wù) - 最終一致性 處理方案！

詳情：?分布式事務(wù)處理方案，微服事務(wù)處理方案

場景（5）

• 消息延時隊列，可做些定時任務(wù)，不固定時間執(zhí)行的定時任務(wù)。
• 例如：用戶下單后如果24小時未支付訂單取消；
• 確認收貨后2天后沒有評價自動好評；
  等...

我們以前的做法是 通常啟用一個定時器，每分鐘或者每小時，去跑一次取出需要處理的訂單或其他數(shù)據(jù)進行處理。
這種做法一個是 效率比較低，如果數(shù)據(jù)量大的話，每次都要掃庫，非常要命！
再者時效性不是很高，最差的時候可能需要等待一輪時長！
還有可能出現(xiàn)重復(fù)執(zhí)行的結(jié)果，時效和輪詢的頻率難以平衡！

利用MQ（Rabbitmq）,DLX （Dead Letter Exchanges）和 消息的 TTL （Time-To-Live Extensions）特性。我們可以高效的完成這個任務(wù)場景！不需要掃庫，時效性更好！

DLX：http://www.rabbitmq.com/dlx.html，
TTL：http://www.rabbitmq.com/ttl.html#per-message-ttl

原理：
發(fā)送到隊列的消息，可以設(shè)置一個存活時間 TTL，在存活時間內(nèi)沒有被消費，可以設(shè)置這個消息轉(zhuǎn)發(fā)到其他隊列里面去；然后我們從這個其他隊列里面消費執(zhí)行我們的任務(wù)，這樣就可以達到一個消息延時的效果！

設(shè)置過期時間：
過期時間可以統(tǒng)一設(shè)置到消息隊列里面，也可以單獨設(shè)置到某個消息！

PS 如果消息設(shè)置了過期時間，發(fā)生到了設(shè)置有過期時間的隊列，已隊列設(shè)置的過期時間為準！

已 SpringBoot 為例：

配置轉(zhuǎn)發(fā)隊列和被轉(zhuǎn)發(fā)隊列：

@Component
@Configuration
public class RabbitMqConfig {
    @Bean
    public Queue curQueue() {
        Map<String, Object> args = new HashMap<String, Object>();
        //超時后的轉(zhuǎn)發(fā)器 過期轉(zhuǎn)發(fā)到 delay_queue_exchange
        args.put("x-dead-letter-exchange", "delay_queue_exchange");
        //routingKey 轉(zhuǎn)發(fā)規(guī)則
        args.put("x-dead-letter-routing-key", "user.#");
        //過期時間 20 秒
        args.put("x-message-ttl", 20000);
        return new Queue("cur_queue", false, false, false, args);
    }
    @Bean
    public Queue delayQueue() {
        return new Queue("delay_queue");
    }
    @Bean
    TopicExchange exchange() {
        //當前隊列
        return new TopicExchange("cur_queue_exchange");
    }
    @Bean
    TopicExchange exchange2() {
        //被轉(zhuǎn)發(fā)的隊列
        return new TopicExchange("delay_queue_exchange");
    }
    @Bean
    Binding bindingHelloQueue(Queue curQueue, TopicExchange exchange) {
         //綁定隊列到轉(zhuǎn)發(fā)器
        return BindingBuilder.bind(curQueue).to(exchange).with("user.#");
    }
    @Bean
    Binding bindingHelloQueue2(Queue delayQueue, TopicExchange exchange2) {
        return BindingBuilder.bind(delayQueue).to(exchange2).with("user.#");
    }
}

發(fā)生消息：

@Component
public class MqEventSender {
    Logger logger = LoggerFactory.getLogger(MqEventSender.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 消息沒有設(shè)置 時間
     *  發(fā)生到隊列 cur_queue_exchange
     * @param msg
     */
    public void sendMsg(String msg) {
        logger.info("發(fā)送消息： " + msg);
        rabbitTemplate.convertAndSend("cur_queue_exchange", "user.ss", msg);
    }
    /**
     * 消息設(shè)置時間
     *  發(fā)生到隊列 cur_queue_exchange
     * @param msg
     */
    public void sendMsgWithTime(String msg) {
        logger.info("發(fā)送消息： " + msg);
        MessageProperties messageProperties = new MessageProperties();
        //過期時間設(shè)置 10 秒
        messageProperties.setExpiration("10000");
        Message message = rabbitTemplate.getMessageConverter().toMessage(msg, messageProperties);
        rabbitTemplate.convertAndSend("cur_queue_exchange", "user.ss", message);
    }
}

消息監(jiān)聽：

監(jiān)聽 的隊列是 delay_queue 而不是 cur_queue；

PS cur_queue 不應(yīng)該有監(jiān)聽者，否則消息被消費達不到想要的延時消息效果！

/**
 * Created by linli on 2017/8/21.
 * 監(jiān)聽 被丟到 超時隊列內(nèi)容
 */
@Component
@RabbitListener(queues = "delay_queue")
public class DelayQueueListener {
    public static Logger logger = LoggerFactory.getLogger(AddCommentsEventListener.class);
    @RabbitHandler
    public void process(@Payload String msg) {
        logger.info("收到消息 "+msg);
    }
}

測試：

/**
 * Created by linli on 2017/8/21.
 */
@RestController
@RequestMapping("/test")
public class TestContorller {
    @Autowired
    MqEventSender sender;
    @RequestMapping("/mq/delay")
    public String test() {
        sender.sendMsg("隊列延時消息!");
        sender.sendMsgWithTime("消息延時消息!");
        return "";
    }
}

結(jié)果：

觀察結(jié)果發(fā)現(xiàn)：發(fā)送時間 和 收到時間 間隔 20秒 ；

我們給消息設(shè)置的 10 秒 TTL 時間沒有生效！驗證了 ： 如果消息設(shè)置了過期時間，發(fā)生到了設(shè)置有過期時間的隊列，已隊列設(shè)置的過期時間為準！

如果希望每個消息都要自己的存活時間，發(fā)送到隊列 不要設(shè)置

args.put(“x-message-ttl”, 20000);

消息的過期時間 設(shè)置在隊列還是消息，根據(jù)自己的業(yè)務(wù)場景去定！

• 總結(jié)

MQ 是一個跨進程的消息隊列，我們可以很好的利用他進行系統(tǒng)的解耦；
引入MQ會給系統(tǒng)帶來一定的復(fù)雜度，需要評估！
MQ 適合做異步任務(wù)，不適合做同步任務(wù)！

文末彩蛋

針對于上面所涉及到的知識點我總結(jié)出了有1到5年開發(fā)經(jīng)驗的程序員在面試中涉及到的絕大部分架構(gòu)面試題及答案做成了文檔和架構(gòu)視頻資料免費分享給大家（包括Dubbo、Redis、Netty、zookeeper、Spring cloud、分布式、高并發(fā)等架構(gòu)技術(shù)資料），希望能幫助到您面試前的復(fù)習(xí)且找到一個好的工作，也節(jié)省大家在網(wǎng)上搜索資料的時間來學(xué)習(xí)，也可以關(guān)注我一下以后會有更多干貨分享。

資料獲取方式 QQ群搜索“708-701-457” 即可免費領(lǐng)取