ps : 這篇文章比較長，讀者還是需要耐心的閱讀的。乾貨多多。

在分散式專案中為了提高效能，也為了實現專案規範，我們都會在處理訊息佇列的時候引入訊息中介軟體。中介軟體的作用一個是為了解耦，還有一個是效能提升。訊息中介軟體我們每個人每天都在接觸，相信大家都用過美團或者是聽過美團。從程式設計師的角度看美團外賣涉及三方角色。【商家】【騎手】【顧客】。這三者的關係簡單理解如下

下面案例會通過程式碼說明，如下是專案結構

rabbit-demo

下訂單

今天我們著重介紹下訂單的流程，那為什麼選擇下訂單流程而不選擇其他兩個流程呢。因為派送流程不是訊息佇列。如果非要規劃為訊息塊我只能認為是訊息消費。而發訂單呢流程和下訂單是一樣的。考慮讀者應該都是平時點餐的那位。對下訂單應該更加的瞭解。

線下買單

我們想想在美團外賣推出之前我們點餐的過程。是顧客到商家店鋪裡進行買單消費的。這種模式存在什麼弊端。這就必須商家的店鋪夠大，人手夠多才能夠讓我們消費者不擁堵起來。但是這樣就增加了商家的成本。所以在這種侷限的條件下我們的線上交易誕生了。

線上交易

有了美團我們就只需要在手機點餐，商家就會進行選擇行接受，然後出貨。這就是下單的一個流程。後面的就是騎手商家流程了。

我們的顧客相當於是在向訊息佇列中新增訊息，訊息體就是我們的訂單。在美團外賣的系統中同一時刻可能多人點單這就導致商家一時間處理不過來，這是我們的mq就作為一中快取機制先將顧客的訂單本地化，mq按先進先出的順序將訊息有條不紊的派發到只能的商家手中。這就解決我們人多導致訂單繁瑣的問題了。

好了以上是我們用mq的好處。下面我們通過程式碼看看mq具體的傳送訊息的集中方式。

mq結構圖

環境準備

rabbitmq安裝

由於RabbitMQ是用Erlang語言編寫的，因此需要先安裝Erlang。

• 通過http://www.erlang.org/downloads獲取對應安裝檔案進行安裝
• 增加環境變數ERLANG_HOME=D:\Program Files\erl9.3(這裡的目錄是我的安裝目錄，你要換成自己的目錄)
• 修改環境變數Path，在原來的值後面加上“;%ERLANG_HOME%\bin”

安裝完Erlang之後，我們就可以安裝RabbitMQ了。

• 到http://www.rabbitmq.com/install-windows-manual.html下載安裝包進行安裝
• 增加環境變數RABBITMQ_HOEM=D:\Program Files\RabbitMQ Server\rabbitmq_server-3.7.5(這裡的目錄是我的安裝目錄，你要換成自己的目錄)
• 修改環境變數Path，在原來的值後面加上“;%RABBITMQ_HOME%\sbin”

安裝好之後，RabbitMQ就作為一個服務按照預設方式進行啟動了

執行命令rabbitmq-plugins enable rabbitmq_management 開啟Web管理外掛

通過瀏覽器訪問http://localhost:15672，並通過預設使用者guest進行登入，密碼也是guest，登入後的頁面：

springboot專案搭建

關於springboot的搭建這裡就不能細說了。我們直接在zxhtom框架中擴充套件了。只要在裡面新增如下依賴就行了

<!-- rabbit mq -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

然後我們只需要建立一個RabbitConfig類。這個類的作用就是讓mq在專案啟動初期檢查並完成各個元件的建立。既然交給spring完成自然需要在類上加上@Configuration註解。

只需要完成以上操作，我們的mq服務就算是配置完成了。剩下的就應該是生產消費了。
這裡比較注意的是RabbitConfig這個類，這個類裡到底是什麼東西，我們在看程式碼前我們先來了解下mq的原理。

rabbitmq元件拆裝

Queue : 訊息佇列 ， 生產者的直接接觸物件。
Exchange ： 交換機 ， 正常情況消費者是直接接觸交換機的。
Routing Key ： 生產者和消費者接觸的兩個不同物件。這就需要routing key 將兩者物件進行關聯起來
Binding ： 通過routing key進行繫結到一起。

下面是消費者、生產者、佇列、交換機、rk、binding之間的關係

四種交換機

上面我們瞭解了mq的內部結構，其實還是很簡單的。但是我們mq有的時候有廣播模式，有的點對點模式。這些是這麼來的呢。對了正是我們本節介紹的內容--交換機。

Direct Exchange

顧名思義就是【直連交換機】。要求傳送的訊息與一個特定的路由鍵完全的匹配。如下程式碼中我們queue1的佇列和directExchange交換機通過DIRECTION繫結到了一起。那麼我們傳送訊息是必須將訊息標記為DIRECTION

@Bean
public Binding binding1(){
    return BindingBuilder.bind(queue1()).to(directExchange()).with(DIRECTIONKEY);
}

public void directionSend() {
    rabbitTemplate.convertAndSend(RabbitConfig.DIRECTEXCHANGE, "directionKey", msg);
}

Fanout Exchange

fanout中文翻譯為【輸出】，在這裡我們理解成廣播。【廣播交換機】
廣播的意思就是所有人都能接受到，在這裡需要重新定義為只要繫結在次交換機上的所有佇列都能接收到。不想其他交換機需要key來進行暗號匹配。Fanout Exchange就是我們常用的廣播模式，只要訂閱就會收到訊息。像一些房產中介就是這種模式，只要你留下手機號碼樓盤動態就會實時的推送在你的手機上。

@Bean
public Binding fanoutBinding1(){
    return BindingBuilder.bind(queue1()).to(fanoutExchange());
}
@Bean
public Binding fanoutBinding2(){
    return BindingBuilder.bind(queue2()).to(fanoutExchange());
}
@Bean
public Binding fanoutBinding3(){
    return BindingBuilder.bind(queue3()).to(fanoutExchange());
}
@Bean
public Binding fanoutBinding4(){
    return BindingBuilder.bind(queue4()).to(fanoutExchange());
}

下面程式碼就是隻要有訊息傳送到Fanout Exchange上所有的佇列就會收到訊息

public void fanoutSend1() {
    rabbitTemplate.convertAndSend(RabbitConfig.FANOUTEXCHANGE, RabbitConfig.QUEUETHIRD, msg);
}

Topic Exchange

將路由鍵和某種模式進行匹配，這裡可以理解成Direction Exchange裡的key可以通過某種正則匹配
# : 表示一個或多個詞，
* : 表示一個詞
zxh.# : 能夠匹配 zxh ; zxh. ; zxh.abc ; zxh.ab.sd.s.a
zxh.* : 能夠匹配 zxh.test ; zxh.demo

Headers Exchange

不處理路由鍵。而是根據傳送的訊息內容中的headers屬性進行匹配。在繫結Queue與Exchange時指定一組鍵值對；當訊息傳送到RabbitMQ時會取到該訊息的headers與Exchange繫結時指定的鍵值對進行匹配；如果完全匹配則訊息會路由到該佇列，否則不會路由到該佇列。headers屬性是一個鍵值對，可以是Hashtable，鍵值對的值可以是任何型別。而fanout，direct，topic 的路由鍵都需要要字串形式的。

匹配規則x-match有下列兩種型別：

x-match = all ：表示所有的鍵值對都匹配才能接受到訊息

x-match = any ：表示只要有鍵值對匹配就能接受到訊息

//請求資料中必須符合headerValues中任意一個引數
@Bean
public Binding headBinding1(){
    Map<String,Object> headerValues = new HashMap<>();
    headerValues.put("type", "cash");
    headerValues.put("aging", "fast");
    return BindingBuilder.bind(queue1()).to(headersExchange()).whereAll(headerValues).match();
}

死信佇列

上面四中交換機介紹完了。我們這裡還有一種叫做死信佇列
還是美團外賣為列， 我們平時美團點單完之後，但是不確定是不是現在就要商家制作。這個時候我們就不會選擇下單。美團就會提示我們30分鐘之內完成付款，負責失效。想想也對美團不可能一直在哪裡等著你下單。你這樣會佔用人家記憶體的。

@Bean
public Queue deadLetterQueue(){
    Map<String, Object> arguments = new HashMap<String, Object>();
    //設定此死信佇列訊息過期後，訊息轉發到那個佇列上
    arguments.put("x-dead-letter-exchange", DIRECTEXCHANGE);
    arguments.put("x-dead-letter-routing-key", DIRECTIONKEY);
    return new Queue(QUEUEDEAD, true, false, false,arguments);
}

@Bean
public Binding deadBinding(){
    return BindingBuilder.bind(deadLetterQueue()).to(directExchange()).with(QUEUEDEAD);
}

public void deadSend() {
    MessagePostProcessor processor = new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            message.getMessageProperties().setExpiration("60000");
            return message;
        }
    };
    log.info("延時訊息開始傳送：");
    rabbitTemplate.convertAndSend(RabbitConfig.DIRECTEXCHANGE, RabbitConfig.QUEUEDEAD, msg,processor);
}

通過上述程式碼我們就能實現傳送一個訊息，會出現延時的效果。這就是我們平時30分鐘支付的效果程式碼。

總結

這篇文章比較長，讀者還是需要耐心的閱讀的。乾貨多多。

加入戰隊

# 加入戰隊

微信公眾號

相關推薦