2. 程式人生 > >RabbitMQ詳解以及和SpringBoot整合

RabbitMQ詳解以及和SpringBoot整合

阿新 • • 發佈：2018-12-12

各訊息元件的適用場景

  • ActiveMQ ActiveMQ 是一款比較早期的訊息元件，由Apache開源出來的，它能滿足吞吐量一般的業務場景，但是對於高併發場景，效能較差。
  • Kafka Kafka追求高吞吐量的特性，它一開始使用於日誌的收集。缺點是訊息可靠性支援較少，適合產生大量資料的網際網路服務的資料收集業務。
  • RocketMQ RocketMQ 早期由阿里團隊開發的，現在升級為Apache的頂級專案。純 Java 開發，具有高吞吐量、高可用性、適合大規模分散式系統應用的特點。缺點是，有些功能不是開源的，如訊息事務。
  • RabbitMQ RabbitMQ 是由 Erlang 語言編寫的，適合對資料一致性、穩定性和可靠性要求很高的場景，對效能和吞吐量的要求還在其次。

AMQP協議

AMQP，即Advanced Message Queuing Protocol，高階訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。 AMQP的主要特徵是面向訊息、佇列、路由（包括點對點和釋出/訂閱）、可靠性、安全。

1、AMQP模型

AMQP協議模型 訊息是直接由生產者傳送到Exchange中的，然後Exchange和Message Queue之間通過某種規則關聯起來（後面會講），生產者是直接訂閱Message Queue的，只要Queue裡面有訊息，就會被消費者消費。這樣就實現了生產者和消費者之間的低耦合性。

2、AMQ核心概念

  • Channel：網路通訊，幾乎所有的操作都在Channel中進行，Channel是進行訊息讀寫的通道。客戶端可建立多個Channel，每個Channel代表一個會話任務。
  • Virtual host：虛擬地址，用於進行邏輯隔離，最上層的訊息路由。
  • Exchange：交換機，接收訊息，根據路由鍵轉發訊息到繫結的佇列。
  • Routing key：一個路由規則，虛擬機器可用它來確定如何路由一個特定訊息
  • Queue：也稱 Message Queue，訊息佇列，儲存訊息，並將它們轉發給消費者。
  • Binding：Exchange和Queue之間的虛擬連線，binding 中可以指定 routing key。

RabbitMQ

RabbitMQ 是基於AMQP協議開發的訊息元件.。

1、RabbitMQ的整體架構

RabbitMQ的整體架構 生產者傳送訊息到Exchange中的，然後Exchange和Message Queue之間通過Routing key建立路由規則，把訊息傳送給特定的Queue，然後訊息推送給訂閱了該Queue的消費者。 RabbitMQ訊息流轉圖

生產者傳送訊息時，需要指定兩個引數，Exchange和Routing key，如果Exchange不指定（即為空），則會採用預設的AMQP default的Exchange，其會根據Routing key的值，路由到對應的 Queue。

2、Exchange

Exchange有四種類型，分別是Direct，Topic，Fanout ，Header。

  • Direct Exchange 所有傳送到Direct Exchange的訊息被轉發到RouteKey中指定的Queue。 注意：Direct模式可以使用RabbitMQ自帶的Exchange:AMQP default，所以不需要將Exchange進行任何繫結（binding）操作，訊息傳遞時，RouteKey必須完全匹配才會被佇列接收，否則該訊息會被拋棄。 exchange在和queue進行binding時會設定routingkey：
channel.QueueBind(queue: "create_pdf_queue",
                    exchange: "pdf_events",
                    routingKey: "pdf_create",
                    arguments: null);

然後我們在將訊息傳送到exchange時會設定對應的routingkey：

channel.BasicPublish(exchange: "pdf_events",
                        routingKey: "pdf_create",
                        basicProperties: properties,
                        body: body);

在direct型別的exchange中，只有這兩個routingkey完全相同，exchange才會選擇對應的binging進行訊息路由。 具體流程如下： Direct 流程

  • Topic Exchange 此型別exchange和上面的direct型別差不多，但direct型別要求routingkey完全相等，這裡的routingkey可以有萬用字元：“*”,“#”,其中 “*” 表示匹配一個單詞， “#”則表示匹配沒有或者多個單詞。 Topic 流程 舉個栗子： Topic Exchange
  • Fanout Exchange 此exchange的路由規則很簡單，直接將訊息路由到所有繫結的佇列中，無須對訊息的routingkey進行匹配操作。 Fanout 流程
  • Header Exchange 此型別的exchange和以上三個都不一樣，其路由的規則是根據header來判斷，其中的header就是以下方法的arguments引數：
Dictionary<string, object> aHeader = new Dictionary<string, object>();
aHeader.Add("format", "pdf");
aHeader.Add("type", "report");
aHeader.Add("x-match", "all");
channel.QueueBind(queue: "queue.A",
                    exchange: "agreements",
                    routingKey: string.Empty,
                    arguments: aHeader);

其中的x-match為特殊的header，可以為all則表示要匹配所有的header，如果為any則表示只要匹配其中的一個header即可。 在釋出訊息的時候就需要傳入header值：

Properties properties = channel.CreateBasicProperties();
properties.Persistent = true;
Dictionary<string, object> mHeader1 = new Dictionary<string, object>();
mHeader1.Add("format", "pdf");
mHeader1.Add("type", "report");
properties.Headers = mHeader1;

以上就是exchange 型別的總結，一般來說direct和topic用來具體的路由訊息，如果要用廣播的訊息一般用fanout的exchange。 header型別用的比較少，但還是知道一點好。

RabbitMQ與SpringBoot整合

  • 新增 amqp 依賴
      <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
      </dependency>
  • 配置檔案 application.properties
# rabbitmq
spring.rabbitmq.host=node2
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
# 對 rabbitmqTemplate 進行監聽,當訊息由於server的原因無法到達queue時，就會被監聽到，以便執行ReturnCallback方法
# 預設為false,Server端會自動刪除不可達訊息
spring.rabbitmq.template.mandatory=true

# 消費端手動確認
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 併發消費
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10
  • 生產者

/**
 * @author K. L. Mao
 * @create 2018/9/20
 */
@Service
public class RabbitSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 用於監聽Server端給我們返回的確認請求,訊息到了exchange，ack 就返回true
     */
    private final RabbitTemplate.ConfirmCallback  confirmCallback = (correlationData, ack, cause) -> {
        System.out.println("correlationData:" + correlationData);
        System.out.println("ack:" + ack);
        if (!ack){
            System.out.println("補償處理...");
        }
    };

    /**
     * 監聽對不可達的訊息進行後續處理;
     * 不可達訊息：指定的路由key路由不到。
     */
    private final RabbitTemplate.ReturnCallback returnCallback = (message, replyCode, replyText,
    exchange, routingKey) -> System.out.println("return exchange:" + exchange + ", routingKey:" + routingKey +
            ", replyText:" + replyText);

    /**
     * 傳送訊息
     * @param order
     */
    public void sendOrder(Order order) {
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
        CorrelationData cd = new CorrelationData();
        // 訊息唯一標識
        cd.setId(UUID.randomUUID().toString().replace("-","") + DateUtils.formatDate(new Date(), "yyyyMMdd"));
        rabbitTemplate.convertAndSend("exchange-2", "springboot.abc", order, cd);
    }
}
  • 消費者
/**
 * @author K. L. Mao
 * @create 2018/9/20
 */
@Service
public class RabbitConsumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "queue-2", durable = "true"),
            exchange = @Exchange(value = "exchange-2",
                    durable = "true", type = "topic",
                    ignoreDeclarationExceptions = "true"),
            key = "springboot.#")
    )
    @RabbitHandler
    public void onOrderMessage(@Payload Order order, @Headers Map<String, Object> properties, Channel channel) throws IOException {
        System.out.println("消費端 order:" + order);
        // deliveryTag: 確認訊息的條數，一般為1
        Long deliveryTag = (Long) properties.get(AmqpHeaders.DELIVERY_TAG);
        System.out.println("deliveryTag:" + deliveryTag);
        // 限流處理：訊息體大小不限制，每次限制消費一條，只作用於該Consumer層，不作用於Channel
        channel.basicQos(0, 1, false);
        // 手工ACK,不批量ack
        channel.basicAck(deliveryTag, false);
    }

可以直接使用@RabbitListener註解，宣告Queue和Exchange以及Binding關係。 消費端接收的訊息是Message物件，結構為：

package org.springframework.messaging;

public interface Message<T> {
    T getPayload();

    MessageHeaders getHeaders();
}

我們可以直接通過註解@Payload 獲取我們傳輸的資料，通過註解@Headers 獲取訊息請求頭。 這裡我們增加了訊息限流的功能，防止生產過多，導致消費者消費吃力的情況： channel.basicQos(0, 1, false)：0表示對訊息的大小無限制，1表示每次只允許消費一條，false表示該限制不作用於channel。 同時，我們這裡採用手工ACK的方式，因為我們配置檔案配置了spring.rabbitmq.listener.simple.acknowledge-mode=manual： channel.basicAck(deliveryTag, false)：deliveryTag表示處理的訊息條數（一般為1），從heaers中取，false表示不批量ack。

DLX

  • DLX定義 DLX為Dead Letter Exchange，死信佇列。當一個訊息在一個佇列中變成死信（dead message）之後，它能重新publish到另一個Exchange，這個Exchange就是DLX。
  • 訊息變成死信的幾種情況 1、訊息被拒絕，ack為false，並且 requeue=false; 2、訊息TTL(Time To Live)過期，指訊息達到了過期時間； 3、佇列達到最大長度。
  • 死信佇列程式碼演示： 1、宣告一個正常的Exchange、Queue以及Binding
    /**
     * 宣告一個正常交換機
     * @return
     */
    @Bean
    public Exchange normalExchange(){
        return ExchangeBuilder.topicExchange("NORMAL_EXCHANGE").durable(true).build();
    }
    /**
     * 宣告一個正常佇列，並且配置正常交換機
     * @return
     */
    @Bean
    public Queue normalQueue(){
        Map<String, Object> args = new HashMap<>(2);
        // x-dead-letter-exchange 宣告 死信交換機
        args.put("x-dead-letter-exchange", DL_EXCHANGE);
        return QueueBuilder.durable("NORMAL_QUEUE").withArguments(args).build();
    }
    /**
     * 繫結，這個是正常的佇列，需要有路由規則
     * @return
     */
    @Bean
    public Binding normalBinding(){
        return BindingBuilder.bind(normalQueue()).to(normalExchange()).with("NORMAL.#").noargs();
    }

2、宣告一個死信佇列，來接收死信訊息

    /**
     * 宣告一個死信交換機,不一定為Topic Exchange, 和交換機型別無關
     * @return
     */
    @Bean
    public Exchange deadLetterExchange(){
        return ExchangeBuilder.topicExchange(DL_EXCHANGE).durable(true).build();
    }
    /**
     * 死信佇列,即requeue、過期(TTL)、超過佇列容量，就會被轉發到此佇列
     * @return
     */
    @Bean
    public Queue deadLetterQueue(){
        return QueueBuilder.durable("DX_QUEUE").build();
    }
    /**
     * 死信佇列進行繫結，一般routingKey為"#"，即所有繫結這個Exchange的Queue都能被路由到
     * @return
     */
    @Bean
    public Binding deadLetterBinding(){
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("#").noargs();
    }

3、對傳送的訊息設定TTL，模擬DLX場景

    /**
     * 傳送死信佇列
     * @param msg
     */
    public void sendDlx(String msg) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        MessagePostProcessor messagePostProcessor = message -> {
            MessageProperties messageProperties = message.getMessageProperties();
            // 設定編碼
            messageProperties.setContentEncoding("UTF-8");
            // 設定過期時間 10s,到了過期時間還沒被消費，則會進入死信佇列
            messageProperties.setExpiration("10000");
            return message;
        };
        rabbitTemplate.convertAndSend("NORMAL_EXCHANGE", "NORMAL.AAA", msg,
                messagePostProcessor, correlationData);
    }

這裡設定了10s的過期時間，訊息一開始是在NORMAL-QUEUE佇列上，如果10s之內還沒被消費，則會進入DX_QUEUE（死信佇列）佇列。

可靠性投遞解決方案

可靠性投遞，即保證訊息的100%被消費。目前，網際網路大廠主要的解決方案，有兩種： 1、訊息落庫，對訊息狀態進行打標 2、訊息的延遲投遞，做二次確認，回撥檢查

  • 訊息落庫 訊息落庫方案
  • 延遲投遞 延遲投遞方案 延遲投遞方案相比訊息落庫方案，優勢是在於把msg db剝離了核心業務，在大業務量的場景中，會減少核心業務的資料庫壓力（少了一次msg db的資料插入）。

相關推薦

RabbitMQ詳解以及和SpringBoot整合

各訊息元件的適用場景 ActiveMQ ActiveMQ 是一款比較早期的訊息元件，由Apache開源出來的，它能滿足吞吐量一般的業務場景，但是對於高併發場景，效能較差。 Kafka Kafka追求高吞吐量的特性，它一開始使用於日誌的收集。缺點是訊息可靠性支援

java實現rabbitMQ延時佇列詳解以及spring-rabbit整合教程

java實現rabbitMQ延時佇列詳解 這是我在公司開發中使用的倆套方案，感興趣的話可以看一下：點選下載 在實際的業務中我們會遇見生產者產生的訊息，不立即消費，而是延時一段時間在消費。RabbitMQ本身沒有直接支援延遲佇列功能，但是我們可以根據其特性Per-Queu

malloc函式詳解以及和new的區別

今天偶然看到一個面試經驗中提到malloc和new的區別，突然發現自己雖然兩個都用過，但是至於區別，真的不是很明白 ，所以就仔細查了一些資料，算是對這個點徹底地瞭解一下，現在把我所學到的記錄下來。 malloc與free是C++/C語言的標準庫函式，new/d

logback日誌元件應用和logback.xml詳解(springmvc和springboot)

本章將通過logback的配置檔案和4個javabean為例項，來講解和驗證logback的使用方式。 最後分別生成debug、info、error類別，以及4個javabean分別生成一個日誌檔案。

Java迭代器（轉）(iterator詳解以及和for迴圈的區別)

迭代器是一種模式，它可以使得對於序列型別的資料結構的遍歷行為與被遍歷的物件分離，即我們無需關心該序列的底層結構是什麼樣子的。只要拿到這個物件,使用迭代器就可以遍歷這個物件的內部.1.IteratorJava提供一個專門的迭代器«interface»Iterator，我們可以對某個序列實現該interface，

Java迭代器(iterator詳解以及和for迴圈的區別)

前言：        迭代器是一種模式、詳細可見其設計模式，可以使得序列型別的資料結構的遍歷行為與被遍歷的物件分離，即我們無需關心該序列的底層結構是什麼樣子的。只要拿到這個物件,使用迭代器就可以遍歷這個物件的內部。Iterable：實現這個介面的集合物件支援迭代，是可以迭代的

ActiveMQ訊息中介軟體 原理詳解 &&附demo實現、以及與springboot整合的demo

一、 訊息中介軟體概述1. 什麼是訊息中介軟體？ 面向訊息的中介軟體(MessageOrlented MiddlewareMOM)較好的解決了以上問題。傳送者將訊息傳送給訊息伺服器， 訊息伺服器將消感

springboot(八)：RabbitMQ詳解

功能 ttr pytho 特征 () png 大量 enc exceptio RabbitMQ 即一個消息隊列，主要是用來實現應用程序的異步和解耦，同時也能起到消息緩沖，消息分發的作用。 消息中間件在互聯網公司的使用中越來越多，剛才還看到新聞阿裏將RocketMQ捐獻給了a

BFS和DFS詳解以及java實現(轉載)

作者： Leo-Yang 原文都先發布在作者個人部落格： http://www.leoyang.net/ 本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連線，否則保留追究法律責任的權利. 前言

SpringMVC與Springfox(Swagger2)整合詳解以及涉及的問題處理

        Swagger 是一個規範和完整的框架，用於生成、描述、呼叫和視覺化 RESTful 風格的 Web 服務。總體目標是使客戶端和檔案系統作為伺服器以同樣的速度來更新。檔案的方法，引數和模型緊密整合到伺服器端的程式碼，允許API來始終保持同步

SpringBoot入門筆記09——屬性配置檔案詳解以及多環境配置

SpringBoot的外部配置 Spring Boot允許將配置外部化（externalize） ，這樣你就能夠在不同的環境下使用相同的程式碼。你可以使用properties檔案，YAML檔案，環境變數和命令列引數來外部化配置。使用@Value註解，可以直接將屬性值注入到beans中，然

詳解swift和OC以及C語言的混編(不看後悔!)

前言:        Swift 語言出來後，可能新的專案直接使用swift來開發，但可能在過程中會遇到一些情況，某些已用OC寫好的類或封裝好的模組，不想再在swift 中再寫一次，或者有一些第三方使用OC寫的,沒有swift版本,怎麼辦?那就使用混編。這個在ＩＯＳ8

Spring學習（1）：控制反轉（IoC）和依賴注入（DI）的詳解以及註解（annotation）開發入門案例

前言 以往的java學習中，我們要想得到一個物件，就把它new出來。如：Apple apple = new Apple(); 在一些複雜的系統中，一個物件A可能依賴於物件B，C等（程式碼表現為A類持有B，C類的物件作為A類的屬性）。以往來說，我們想要使用B，

RabbitMQ的安裝及和springboot的整合

Message Broker與AMQP簡介 Message Broker是一種訊息驗證、傳輸、路由的架構模式，其設計目標主要應用於下面這些場景： 訊息路由到一個或多個目的地 訊息轉化為其他的表現方式 執行訊息的聚集、訊息的分解，並將結果傳送到他們的目的地，然後重新組合

詳解html和thymeleaf中的相對路徑，解決springboot前臺頁面的相對路徑問題

一.問題 最近在使用springboot框架，眾所周知，該框架可以直接以jar的方式執行，在該方式執行的情況下，預設contextPath是/。在前臺頁面引用的時候，我是這樣寫相對路徑的: 後臺Req

JSON詳解以及stringfy( )和parse( )的使用

前言：最早，XML是是網際網路上傳輸結構化資料的事實標準，但是XML使用繁瑣，為了解決這個問題，JSON應運而生。JSON是JavaScript的一個嚴格子集，利用了 JavaScript中的一些模式來表示結構化資料。 JSON 並不從屬於 JavaScript

BFS和DFS詳解以及java實現

前言 圖在演算法世界中的重要地位是不言而喻的，曾經看到一篇Google的工程師寫的一篇《Get that job at Google!》文章中說到面試官問的問題中幾乎有一半的問題都可以用圖的方法去解決。由此也可以看出圖確實適用範圍確實很廣。 圖的表示 閒話不多說，首先要

TCP三次握手和四次揮手詳解以及原因

三次握手 三次握手（three times handshake；three-way handshake）所謂的“三次握手”即對每次傳送的資料量是怎樣跟蹤進行協商使資料段的傳送和接收同步，根據所接收到的資料量而確定的資料確認數及資料傳送、接收完畢後何時撤消聯絡，

python 序列型別以及list(列表) 和 tuple(元組)詳解 以及range的用法

參考： #################################################################### 序列：成員是有序排列，可以通過下標偏移量訪問到它的一個或多個成員 在python中有六種內建的序列：列表、元素、字串、

TCP通訊過程詳解以及tcp長連線和短連線

1. TCP連線 當網路通訊時採用TCP協議時，在真正的讀寫操作之前，server與client之間必須建立一個連線，當讀寫操作完成後，雙方不再需要這個連線 時它們可以釋放這個連線，連線的建立是需要三次握手的，而釋放則需要4次揮手，所以說每個連線的建立都是需要