服務(wù)異步通信-分布式場景下的應(yīng)用

如果單機模式忘記也可以看看這個快速回顧rabbitmq,在做學(xué)習(xí)

消息隊列在使用過程中，面臨著很多實際問題需要思考：

1.消息可靠性

消息從發(fā)送，到消費者接收，會經(jīng)理多個過程：

其中的每一步都可能導(dǎo)致消息丟失，常見的丟失原因包括：

• 發(fā)送時丟失：
  • 生產(chǎn)者發(fā)送的消息未送達exchange
  • 消息到達exchange后未到達queue
• MQ宕機，queue將消息丟失
• consumer接收到消息后未消費就宕機

針對這些問題，RabbitMQ分別給出了解決方案：

• 生產(chǎn)者確認(rèn)機制
• mq持久化
• 消費者確認(rèn)機制
• 失敗重試機制

下面我們就通過案例來演示每一個步驟。
該工程一個消息生產(chǎn)者

一個消息監(jiān)聽者

用到依賴

  <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!--AMQP依賴，包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!--單元測試-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
    </dependencies>

配置文件生成一個信道一會演示

@Configuration
public class CommonConfig {
    @Bean
   public Queue SimpleQueue(){
        return new Queue("simple.queue");
    }
}

1.1.生產(chǎn)者消息確認(rèn)

RabbitMQ提供了publisher confirm機制來避免消息發(fā)送到MQ過程中丟失。這種機制必須給每個消息指定一個唯一ID。消息發(fā)送到MQ以后，會返回一個結(jié)果給發(fā)送者，表示消息是否處理成功。

返回結(jié)果有兩種方式：

• publisher-confirm，發(fā)送者確認(rèn)
  • 消息成功投遞到交換機，返回ack
  • 消息未投遞到交換機，返回nack
• publisher-return，發(fā)送者回執(zhí)
  • 消息投遞到交換機了，但是沒有路由到隊列。返回ACK，及路由失敗原因。

注意：

1.1.1.修改配置實現(xiàn)

首先，修改publisher服務(wù)中的application.yml文件，添加下面的內(nèi)容：

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
   

說明：

• publish-confirm-type：開啟publisher-confirm，這里支持兩種類型：
  • simple：同步等待confirm結(jié)果，直到超時
  • correlated：異步回調(diào)，定義ConfirmCallback，MQ返回結(jié)果時會回調(diào)這個ConfirmCallback
• publish-returns：開啟publish-return功能，同樣是基于callback機制，不過是定義ReturnCallback
• template.mandatory：定義消息路由失敗時的策略。true，則調(diào)用ReturnCallback；false：則直接丟棄消息

1.1.2.定義Return回調(diào)

每個RabbitTemplate只能配置一個（消息回調(diào)）ReturnCallback，因此需要在項目加載時配置：

• ReturnCallback: 處理消息是否被正確路由到隊列，通常在消息無法被路由時觸發(fā)。
• ConfirmCallback: 處理消息是否成功發(fā)送到Broker，無論消息是否被正確路由到隊列。它是消息成功投遞到Broker的確認(rèn)機制。

擴展

ReturnCallback（消息退回機制）：

ReturnCallback 用于處理消息被Broker(交換機)退回的情況。這發(fā)生在消息發(fā)送時，消息無法被正確路由到任何隊列時（通常是因為沒有匹配的隊列綁定到消息交換機）。
當(dāng)消息被退回時，ReturnCallback 中的方法會被觸發(fā)，允許你處理或記錄未被正確路由的消息。

ReturnCallback 是在消息發(fā)送之前設(shè)置的，通常用于處理發(fā)送不可路由消息的情況。如果消息成功路由到隊列，它不會被觸發(fā)。

ConfirmCallback（消息發(fā)送者的失敗回調(diào)）：

ConfirmCallback 用于處理消息是否成功發(fā)送到Broker的確認(rèn)情況。它通常用于確認(rèn)消息是否被成功投遞到Broker，即使消息已被正確路由到隊列。
ConfirmCallback 在消息發(fā)送后，Broker返回確認(rèn)信息時觸發(fā)。確認(rèn)信息可以表示消息已成功發(fā)送到Broker（ACK）或發(fā)送到Broker后發(fā)生錯誤（NACK）。

ConfirmCallback 不涉及消息是否被正確路由到隊列，而是關(guān)注消息是否成功發(fā)送到Broker。

修改publisher服務(wù)，添加一個：

package cn.itcast.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 獲取RabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        // 設(shè)置ReturnCallback
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            // 投遞失敗，記錄日志
            log.info("消息發(fā)送失敗，應(yīng)答碼{}，原因{}，交換機{}，路由鍵{},消息{}",
                     replyCode, replyText, exchange, routingKey, message.toString());
            // 如果有業(yè)務(wù)需要，可以重發(fā)消息
        });
    }
}

returncallback源碼

測試之前對交換機,路由key,隊列進行綁定

@Configuration
public class CommonConfig {

    @Bean
    public Queue SimpleQueue(){
        return new Queue("simple.queue");
    }

    /**
     * 交換機以及隊列綁定
     */
    @Bean
    TopicExchange exchange(){
        return new TopicExchange("amq.topic");
    }
    @Bean
    Binding binding(){
        return BindingBuilder.bind(SimpleQueue()).to(exchange()).with("simple.test");
    }

}

1.1.3.定義ConfirmCallback

ConfirmCallback可以在發(fā)送消息時指定，因為每個業(yè)務(wù)處理confirm成功或失敗的邏輯不一定相同。

在publisher服務(wù)的cn.itcast.mq.spring.SpringAmqpTest類中，定義一個單元測試方法：

 @Test
    public void testSendMessage2SimpleQueue() throws InterruptedException {
        // 1.消息體
        String message = "hello, spring amqp!";
        // 2.全局唯一的消息ID，需要封裝到CorrelationData中
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 3.添加callback
        correlationData.getFuture().addCallback(
                result -> {
                    if(result.isAck()){
                        // 3.1.ack，消息成功
                        log.debug("消息發(fā)送成功, ID:{}", correlationData.getId());
                    }else{
                        // 3.2.nack，消息失敗
                        log.error("消息發(fā)送失敗, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                    }
                },
                ex -> log.error("消息發(fā)送異常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
        );
        // 4.發(fā)送消息
        rabbitTemplate.convertAndSend("amq.topic", "simple.test", message, correlationData);

      
    }

rabitmqtemplate發(fā)消息的接口添加參數(shù)

• correlationData 包含消息id和回調(diào)

可以定義倆個callback

測試結(jié)果

1.2.消息持久化

生產(chǎn)者確認(rèn)可以確保消息投遞到RabbitMQ的隊列中，但是消息發(fā)送到RabbitMQ以后，如果突然宕機，也可能導(dǎo)致消息丟失。

要想確保消息在RabbitMQ中安全保存，必須開啟消息持久化機制。

• 交換機持久化
• 隊列持久化
• 消息持久化

1.2.1.交換機持久化

RabbitMQ中交換機默認(rèn)是非持久化的，mq重啟后就丟失。

SpringAMQP中可以通過代碼指定交換機持久化：

@Bean
public DirectExchange simpleExchange(){
    // 三個參數(shù)：交換機名稱、是否持久化、當(dāng)沒有queue與其綁定時是否自動刪除
    return new DirectExchange("simple.direct", true, false);
}

事實上，默認(rèn)情況下，由SpringAMQP聲明的交換機都是持久化的。

可以在RabbitMQ控制臺看到持久化的交換機都會帶上D的標(biāo)示：

查看配置文件注入時候的源碼,其實都是默認(rèn)持久化的

1.2.2.隊列持久化

RabbitMQ中隊列默認(rèn)是非持久化的，mq重啟后就丟失。

SpringAMQP中可以通過代碼指定交換機持久化：

@Bean
public Queue simpleQueue(){
    // 使用QueueBuilder構(gòu)建隊列，durable就是持久化的
    return QueueBuilder.durable("simple.queue").build();
}

事實上，默認(rèn)情況下，由SpringAMQP聲明的隊列都是持久化的。

可以在RabbitMQ控制臺看到持久化的隊列都會帶上D的標(biāo)示：

交換機和隊列在進行注入的時候就默認(rèn)是持久化,但是消息并不是持久化的需要進行指定

1.2.3.消息持久化

利用SpringAMQP發(fā)送消息時，可以設(shè)置消息的屬性（MessageProperties），指定delivery-mode：

• 1：非持久化
• 2：持久化

用java代碼指定：

默認(rèn)情況下，SpringAMQP發(fā)出的任何消息都是持久化的，不用特意指定。

發(fā)送消息

此時消費端是關(guān)閉的,所以可以在隊列進行查看消息

重啟docker 部署的mq

docker restart mq

消息存在

1.3.消費者消息確認(rèn)

RabbitMQ是閱后即焚機制，RabbitMQ確認(rèn)消息被消費者消費后會立刻刪除。

而RabbitMQ是通過消費者回執(zhí)來確認(rèn)消費者是否成功處理消息的：消費者獲取消息后，應(yīng)該向RabbitMQ發(fā)送ACK回執(zhí)，表明自己已經(jīng)處理消息。

設(shè)想這樣的場景：

• 1）RabbitMQ投遞消息給消費者
• 2）消費者獲取消息后，返回ACK給RabbitMQ
• 3）RabbitMQ刪除消息
• 4）消費者宕機，消息尚未處理

這樣，消息就丟失了。因此消費者返回ACK的時機非常重要。

而SpringAMQP則允許配置三種確認(rèn)模式：

?manual：手動ack，需要在業(yè)務(wù)代碼結(jié)束后，調(diào)用api發(fā)送ack。

?auto：自動ack，由spring監(jiān)測listener代碼是否出現(xiàn)異常，沒有異常則返回ack；拋出異常則返回nack

?none：關(guān)閉ack，MQ假定消費者獲取消息后會成功處理，因此消息投遞后立即被刪除

由此可知：

• none模式下，消息投遞是不可靠的，可能丟失
• auto模式類似事務(wù)機制，出現(xiàn)異常時返回nack，消息回滾到mq；沒有異常，返回ack
• manual：自己根據(jù)業(yè)務(wù)情況，判斷什么時候該ack

一般，我們都是使用默認(rèn)的auto即可。

1.3.1.演示none模式

修改consumer服務(wù)的application.yml文件，添加下面內(nèi)容：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: none # 關(guān)閉ack

修改consumer服務(wù)的SpringRabbitListener類中的方法，模擬一個消息處理異常：

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
    log.info("消費者接收到simple.queue的消息：【{}】", msg);
    // 模擬異常 
    System.out.println(1 / 0);
    log.debug("消息處理完成！");
}

測試可以發(fā)現(xiàn)，當(dāng)消息處理拋異常時，消息依然被RabbitMQ刪除了。

消息被刪除

也就是說不關(guān)消息設(shè)計的業(yè)務(wù)是否正常執(zhí)行,當(dāng)消費者回到后該模式就直接刪除消息

1.3.2.演示auto模式

再次把確認(rèn)機制修改為auto:

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto # 關(guān)閉ack

在異常位置打斷點，再次發(fā)送消息，程序卡在斷點時，可以發(fā)現(xiàn)此時消息狀態(tài)為unack（未確定狀態(tài)）：

拋出異常后，因為Spring會自動返回nack，所以消息恢復(fù)至Ready狀態(tài)，并且沒有被RabbitMQ刪除：

1.4.消費失敗重試機制

當(dāng)消費者出現(xiàn)異常后，消息會不斷requeue（重入隊）到隊列，再重新發(fā)送給消費者，然后再次異常，再次requeue，無限循環(huán)，導(dǎo)致mq的消息處理飆升，帶來不必要的壓力：

會不斷的重新投遞重新消息

怎么辦呢？

1.4.1.本地重試

可以利用Spring的retry機制，在消費者出現(xiàn)異常時利用本地重試，而不是無限制的requeue到mq隊列。

修改consumer服務(wù)的application.yml文件，添加內(nèi)容：

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 開啟消費者失敗重試
          initial-interval: 1000 # 初識的失敗等待時長為1秒
          multiplier: 1 # 失敗的等待時長倍數(shù)，下次等待時長 = multiplier * last-interval
          max-attempts: 3 # 最大重試次數(shù)
          stateless: true # true無狀態(tài)；false有狀態(tài)。如果業(yè)務(wù)中包含事務(wù)，這里改為false

重啟consumer服務(wù)，重復(fù)之前的測試?？梢园l(fā)現(xiàn)：

• 在重試3次后，SpringAMQP會拋出異常AmqpRejectAndDontRequeueException，說明本地重試觸發(fā)了
• 查看RabbitMQ控制臺，發(fā)現(xiàn)消息被刪除了，說明最后SpringAMQP返回的是ack，mq刪除消息了

結(jié)論：

• 開啟本地重試時，消息處理過程中拋出異常，不會requeue到隊列，而是在消費者本地重試
• 重試達到最大次數(shù)后，Spring會返回ack，消息會被丟棄

1.4.2.失敗策略

在之前的測試中，達到最大重試次數(shù)后，消息會被丟棄，這是由Spring內(nèi)部機制決定的。

在開啟重試模式后，重試次數(shù)耗盡，如果消息依然失敗，則需要有MessageRecovery接口來處理，它包含三種不同的實現(xiàn)：

• RejectAndDontRequeueRecoverer：重試耗盡后，直接reject，丟棄消息。默認(rèn)就是這種方式

• ImmediateRequeueMessageRecoverer：重試耗盡后，返回nack，消息重新入隊 （如果繼續(xù)出現(xiàn)異常繼續(xù)按照此循環(huán)）

• RepublishMessageRecoverer：重試耗盡后，將失敗消息投遞到指定的交換機

比較優(yōu)雅的一種處理方案是**RepublishMessageRecoverer**，失敗后將消息投遞到一個指定的，專門存放異常消息的隊列，后續(xù)由人工集中處理。

1）在consumer服務(wù)中定義處理失敗消息的交換機和隊列

@Bean
public DirectExchange errorMessageExchange(){
    return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){
    return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
    return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

2）定義一個RepublishMessageRecoverer，關(guān)聯(lián)隊列和交換機

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

完整代碼：

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;

@Configuration
public class ErrorMessageConfig {
    
    /**
     * 定義統(tǒng)一處理異常消息的隊列和交換機
     *
     */
    @Bean
    public Queue ErrorQueue(){
        return new Queue("errorQueue");
    }
    @Bean
    public DirectExchange ErrorExchange(){
        return new DirectExchange("errorExchange");
    }
    @Bean
    public Binding errorBinding(){
        return BindingBuilder.bind(ErrorQueue()).to(ErrorExchange()).with("error");
    }
//異常消息處理器
	  @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
    //失敗的消息就會發(fā)送到這個routingkey
        return new RepublishMessageRecoverer(rabbitTemplate, "errorExchange", "error");
    }
}

啟動 查看bing關(guān)系

測試
發(fā)送消息

日志輸出
進行了三次本地重試,然后轉(zhuǎn)發(fā)給了定義的error隊列

在error隊列中可以看到消息體和消息異常轉(zhuǎn)發(fā)原因

然后可以通過監(jiān)聽這個錯誤消息隊列來進行人工處理,是整個異步通信中的異常兜底方案

1.5.總結(jié)

如何確保RabbitMQ消息的可靠性？

• 開啟生產(chǎn)者確認(rèn)機制，確保生產(chǎn)者的消息能到達隊列
• 開啟持久化功能，確保消息未消費前在隊列中不會丟失
• 開啟消費者確認(rèn)機制為auto，由spring確認(rèn)消息處理成功后完成ack
• 開啟消費者失敗重試機制，并設(shè)置MessageRecoverer，多次重試失敗后將消息投遞到異常交換機，交由人工處理

2.死信交換機

2.1.初識死信交換機

2.1.1.什么是死信交換機

什么是死信？

當(dāng)一個隊列中的消息滿足下列情況之一時，可以成為死信（dead letter）：

• 消費者使用basic.reject或 basic.nack聲明消費失敗，并且消息的requeue參數(shù)設(shè)置為false
• 消息是一個過期消息，超時無人消費
• 要投遞的隊列消息滿了，無法投遞

如果這個包含死信的隊列配置了dead-letter-exchange屬性，指定了一個交換機，那么隊列中的死信就會投遞到這個交換機中，而這個交換機稱為死信交換機（Dead Letter Exchange，檢查DLX）。

如圖，一個消息被消費者拒絕了，變成了死信：

因為simple.queue綁定了死信交換機 dl.direct，因此死信會投遞給這個交換機：

如果這個死信交換機也綁定了一個隊列，則消息最終會進入這個存放死信的隊列：

另外，隊列將死信投遞給死信交換機時，必須知道兩個信息：

• 死信交換機名稱
• 死信交換機與死信隊列綁定的RoutingKey

這樣才能確保投遞的消息能到達死信交換機，并且正確的路由到死信隊列。

2.1.2.利用死信交換機接收死信（拓展）

在失敗重試策略中，默認(rèn)的RejectAndDontRequeueRecoverer會在本地重試次數(shù)耗盡后，發(fā)送reject給RabbitMQ，消息變成死信，被丟棄。

我們可以給simple.queue添加一個死信交換機，給死信交換機綁定一個隊列。這樣消息變成死信后也不會丟棄，而是最終投遞到死信交換機，路由到與死信交換機綁定的隊列。

我們在consumer服務(wù)中，定義一組死信交換機、死信隊列：

// 聲明普通的 simple.queue隊列，并且為其指定死信交換機：dl.direct
@Bean
public Queue simpleQueue2(){
    return QueueBuilder.durable("simple.queue") // 指定隊列名稱，并持久化
        .deadLetterExchange("dl.direct") // 指定死信交換機
        .build();
}
// 聲明死信交換機 dl.direct
@Bean
public DirectExchange dlExchange(){
    return new DirectExchange("dl.direct", true, false);
}
// 聲明存儲死信的隊列 dl.queue
@Bean
public Queue dlQueue(){
    return new Queue("dl.queue", true);
}
// 將死信隊列 與 死信交換機綁定
@Bean
public Binding dlBinding(){
    return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
}

2.1.3.總結(jié)

什么樣的消息會成為死信？

• 消息被消費者reject或者返回nack
• 消息超時未消費
• 隊列滿了

死信交換機的使用場景是什么？

• 如果隊列綁定了死信交換機，死信會投遞到死信交換機；
• 可以利用死信交換機收集所有消費者處理失敗的消息（死信），交由人工處理，進一步提高消息隊列的可靠性。

死信交換機和專門設(shè)置交換機和隊列做異常處理機制的哪個不同的是,異常處理是消費者出現(xiàn)業(yè)務(wù)異常,消費方只能把該消息發(fā)送給專門的處理隊列,而死信交換機則是由隊列處理無效的消息交給專門的交換機-隊列,也是一種安全兜底方案

2.2.TTL

一個隊列中的消息如果超時未消費，則會變?yōu)樗佬?，超時分為兩種情況：

• 消息所在的隊列設(shè)置了超時時間
• 消息本身設(shè)置了超時時間

所以延遲消息的原理就是在對一個交換機和隊列做ttl時間限制,但是沒有消費者消費該隊列的消息,所以超過時間的消息都會轉(zhuǎn)發(fā)給死信交換機,消費者對死信交換機進行消費,就達到了延遲消息的效果

2.2.1.接收超時死信的死信交換機

在consumer服務(wù)的SpringRabbitListener中，定義一個新的消費者，并且聲明 死信交換機、死信隊列：

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "dl.ttl.queue", durable = "true"),
    exchange = @Exchange(name = "dl.ttl.direct"),
    key = "ttl"
))
public void listenDlQueue(String msg){
    log.info("接收到 dl.ttl.queue的延遲消息：{}", msg);
}

2.2.2.聲明一個隊列，并且指定TTL

要給隊列設(shè)置超時時間，需要在聲明隊列時配置x-message-ttl屬性：

@Bean
public Queue ttlQueue(){
    return QueueBuilder.durable("ttl.queue") // 指定隊列名稱，并持久化
        .ttl(10000) // 設(shè)置隊列的超時時間，10秒
        .deadLetterExchange("dl.ttl.direct") // 指定死信交換機
        .build();
}

注意，這個隊列設(shè)定了死信交換機為dl.ttl.direct

聲明交換機，將ttl與交換機綁定：

@Bean
public DirectExchange ttlExchange(){
    return new DirectExchange("ttl.direct");
}
@Bean
public Binding ttlBinding(){
    return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

此時就申明了一個交換機和隊列,并且隊列指定了超時時間,

發(fā)送消息，但是不要指定TTL：

@Test
public void testTTLQueue() {
    // 創(chuàng)建消息
    String message = "hello, ttl queue";
    // 消息ID，需要封裝到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 發(fā)送消息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    // 記錄日志
    log.debug("發(fā)送消息成功");
}

發(fā)送消息的日志：

查看下接收消息的日志：

因為隊列的TTL值是10000ms，也就是10秒?？梢钥吹较l(fā)送與接收之間的時差剛好是10秒。

2.2.3.發(fā)送消息時，設(shè)定TTL

在發(fā)送消息時，也可以指定TTL：

@Test
public void testTTLMsg() {
    // 創(chuàng)建消息
    Message message = MessageBuilder
        .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))
        .setExpiration("5000")
        .build();
    // 消息ID，需要封裝到CorrelationData中
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    // 發(fā)送消息
    rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);
    log.debug("發(fā)送消息成功");
}

這次，發(fā)送與接收的延遲只有5秒。說明當(dāng)隊列、消息都設(shè)置了TTL時，任意一個到期就會成為死信。
消息會以短的為準(zhǔn),只要超時就是私信

2.2.4.總結(jié)

消息超時的兩種方式是？

• 給隊列設(shè)置ttl屬性，進入隊列后超過ttl時間的消息變?yōu)樗佬?/li>
• 給消息設(shè)置ttl屬性，隊列接收到消息超過ttl時間后變?yōu)樗佬?/li>

如何實現(xiàn)發(fā)送一個消息20秒后消費者才收到消息？

• 給消息的目標(biāo)隊列指定死信交換機
• 將消費者監(jiān)聽的隊列綁定到死信交換機
• 發(fā)送消息時給消息設(shè)置超時時間為20秒

2.3.延遲隊列

利用TTL結(jié)合死信交換機，我們實現(xiàn)了消息發(fā)出后，消費者延遲收到消息的效果。這種消息模式就稱為延遲隊列（Delay Queue）模式。

延遲隊列的使用場景包括：

• 延遲發(fā)送短信
• 用戶下單，如果用戶在15 分鐘內(nèi)未支付，則自動取消
• 預(yù)約工作會議，20分鐘后自動通知所有參會人員

因為延遲隊列的需求非常多，所以RabbitMQ的官方也推出了一個插件，原生支持延遲隊列效果。

這個插件就是DelayExchange插件。參考RabbitMQ的插件列表頁面：https://www.rabbitmq.com/community-plugins.html

使用方式可以參考官網(wǎng)地址：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq

2.3.1.安裝DelayExchange插件

docker 容器部署需要把插件上傳到容器內(nèi)部,所以先從新創(chuàng)建插件文件夾或者數(shù)據(jù)卷進行掛載

 docker run -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123 -v mq-plugins:/plugins  --name mq  -p 15672:15672  -p 5672:5672  -d  rabbitmq:3.8-management

RabbitMQ有一個官方的插件社區(qū)，地址為：https://www.rabbitmq.com/community-plugins.html

下載后復(fù)制到掛載的宿主機掛載目錄
查詢掛載的數(shù)據(jù)卷在數(shù)組機位置

docker volume inspect mq-plugins

將下載好的插件復(fù)制出去

sudo cp rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez  /var/lib/docker/volumes/mq-plugins/_data

進入容器

輸入命令

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

運行成功
此時在web頁面添加交換機時,就可以使用死信交換機

2.3.2.DelayExchange原理

DelayExchange需要將一個交換機聲明為delayed類型。當(dāng)我們發(fā)送消息到delayExchange時，流程如下：

• 接收消息
• 判斷消息是否具備x-delay屬性
• 如果有x-delay屬性，說明是延遲消息，持久化到硬盤，讀取x-delay值，作為延遲時間
• 返回routing not found結(jié)果給消息發(fā)送者
• x-delay時間到期后，重新投遞消息到指定隊列

2.3.3.使用DelayExchange

插件的使用也非常簡單：聲明一個交換機，交換機的類型可以是任意類型，只需要設(shè)定delayed屬性為true即可，然后聲明隊列與其綁定即可。

1）聲明DelayExchange交換機

基于注解方式（推薦）：

   @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue", durable = "true"),
            exchange = @Exchange(name = "delay.direct",delayed = "true"),
            key = "delay"
    ))
    public  void ListernDeadExchange(String msg) {
        log.info("接收到 延遲交換機的延遲消息：{}", msg);
    }

也可以基于@Bean的方式：

2）發(fā)送消息

發(fā)送消息時，一定要攜帶(消息頭)x-delay屬性，指定延遲的時間：

測試代碼
生產(chǎn)者

  @Test
    public void testDelayQueue() {
        // 創(chuàng)建消息
        Message message = MessageBuilder.withBody("hello, delay queue".getBytes(StandardCharsets.UTF_8))
                .setHeader("x-delay", 5000).build();
        // 消息ID，需要封裝到CorrelationData中
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        correlationData.getFuture().addCallback(
                result -> {
                    if(result.isAck()){
                        // 3.1.ack，消息成功
                        log.debug("消息發(fā)送成功, ID:{}", correlationData.getId());
                    }else{
                        // 3.2.nack，消息失敗
                        log.error("消息發(fā)送失敗, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                    }
                },
                ex -> log.error("消息發(fā)送異常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
        );
        // 發(fā)送消息
        rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
        // 記錄日志
        log.debug("發(fā)送消息成功");
    }

消費者
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue", durable = "true"),
            exchange = @Exchange(name = "delay.direct",delayed = "true"),
            key = "delay"
    ))
    public  void ListernDeadExchange(String msg) {
        log.info("接收到 延遲交換機的延遲消息：{}", msg);
    }

消費者5s后成功收到消息

生產(chǎn)者而報錯沒有路由,明明代碼寫了路由

原因是消息的退回異常路由處理,沒有異常處理轉(zhuǎn)發(fā)而是放在隊列里進行報存5s,沒有直接到達消費者

雖然業(yè)務(wù)實現(xiàn)了,但是一直占用退回處理機制會對服務(wù)資源進行消耗所以這里進行修改退回回調(diào)

 @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
//        通過上下文對象獲取注入到spring ioc的bean對象進行加工處理
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);

        /**
         * message – the returned message.
         * replyCode – the reply code.
         * replyText – the reply text.
         * exchange – the exchange. routingKey – the routing key.
         */
        rabbitTemplate.setReturnCallback((message, code, text, exchange, routkey) -> {
            //判斷是否是延遲消息
            Integer delay = message.getMessageProperties().getReceivedDelay();
            if (delay>0){
                return ;
            }


            log.error("被退回消息，響應(yīng)碼:{},響應(yīng)信息:{},目標(biāo){}，路由key{}", code, text, exchange, routkey);
            //然后就是處理邏輯
            System.out.println("開發(fā)人員處理");

        });
    }

2.3.4.總結(jié)

延遲隊列插件的使用步驟包括哪些？

?聲明一個交換機，添加delayed屬性為true

?發(fā)送消息時，添加x-delay頭，值為超時時間

3.惰性隊列

3.1.消息堆積問題

當(dāng)生產(chǎn)者發(fā)送消息的速度超過了消費者處理消息的速度，就會導(dǎo)致隊列中的消息堆積，直到隊列存儲消息達到上限。之后發(fā)送的消息就會成為死信，默認(rèn)會被丟棄，這就是消息堆積問題。

解決消息堆積有兩種思路：

• 增加更多消費者，提高消費速度。也就是我們之前說的work queue模式
• 擴大隊列容積，提高堆積上限

要提升隊列容積，把消息保存在內(nèi)存中顯然是不行的,內(nèi)存存儲的服務(wù),一旦涉及到高并發(fā),安全性無法得到保證,容易產(chǎn)生宕機的風(fēng)險

3.2.惰性隊列

從RabbitMQ的3.6.0版本開始，就增加了Lazy Queues的概念，也就是惰性隊列。惰性隊列的特征如下：

• 接收到消息后直接存入磁盤而非內(nèi)存
• 消費者要消費消息時才會從磁盤中讀取并加載到內(nèi)存
• 支持?jǐn)?shù)百萬條的消息存儲

這樣的讀取比起原生肯定所有下降的,但是消息量大到生產(chǎn)>消費還是挺難的,需要酌情考慮

3.2.1.基于命令行設(shè)置lazy-queue

而要設(shè)置一個隊列為惰性隊列，只需要在聲明隊列時，指定x-queue-mode屬性為lazy即可?？梢酝ㄟ^命令行將一個運行中的隊列修改為惰性隊列：

rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues  

命令解讀：

• rabbitmqctl ：RabbitMQ的命令行工具
• set_policy ：添加一個策略
• Lazy ：策略名稱，可以自定義
• "^lazy-queue$" ：用正則表達式匹配隊列的名字
• '{"queue-mode":"lazy"}' ：設(shè)置隊列模式為lazy模式
• --apply-to queues ：策略的作用對象，是所有的隊列

3.2.2.基于@Bean聲明lazy-queue

無論是queue,還是交換機,或者消息,需要涉及到第三方的附加業(yè)務(wù)時候需要構(gòu)造模式進行加工

3.2.3.基于@RabbitListener聲明LazyQueue

測試;

    @Bean 
    public Queue lazyQueue(){
        return  QueueBuilder.durable("lazy.queue")
                .lazy()
                .build();
    } 
    @Bean 
    public Queue commonQueue(){
        return  QueueBuilder.durable("common.queue")
                .lazy()
                .build();
    }

聲明一個普通隊列一個懶隊列
發(fā)送消息

 @Test
    public void testLazyQueue() {
        // 創(chuàng)建消息  設(shè)置為非持久化
        Message message = MessageBuilder.withBody("hello, lazy queue".getBytes(StandardCharsets.UTF_8))
                .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();

        // 消息ID，需要封裝到CorrelationData中
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        correlationData.getFuture().addCallback(
                result -> {
                    if(result.isAck()){
                        // 3.1.ack，消息成功
                        log.debug("消息發(fā)送成功, ID:{}", correlationData.getId());
                    }else{
                        // 3.2.nack，消息失敗
                        log.error("消息發(fā)送失敗, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                    }
                },
                ex -> log.error("消息發(fā)送異常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
        );
        // 發(fā)送消息

//       for (int i = 0; i < 100000; i++) {
//                rabbitTemplate.convertAndSend("lazy.queue", message, correlationData);
//            }
   for (int i = 0; i < 100000; i++) {
                rabbitTemplate.convertAndSend("common.queue", message, correlationData);
            }




    }

common普通：隊列開始接收消息

接收完畢 消息全是寫在內(nèi)存 會占用內(nèi)存較多

lazy隊列:
正常收到消息

發(fā)送完畢后查看

消息無丟失,并且都在磁盤

3.3.總結(jié)

消息堆積問題的解決方案？

• 隊列上綁定多個消費者，提高消費速度
• 使用惰性隊列，可以再mq中保存更多消息

惰性隊列的優(yōu)點有哪些？

• 基于磁盤存儲，消息上限高
• 沒有間歇性的page-out，性能比較穩(wěn)定

惰性隊列的缺點有哪些？

• 基于磁盤存儲，消息時效性會降低
• 性能受限于磁盤的IO

4.MQ集群

4.1.集群分類

RabbitMQ的是基于Erlang語言編寫，而Erlang又是一個面向并發(fā)的語言，天然支持集群模式。RabbitMQ的集群有兩種模式：

?普通集群：是一種分布式集群，將隊列分散到集群的各個節(jié)點，從而提高整個集群的并發(fā)能力,但是該普通集群模式一但結(jié)點丟失就會丟失數(shù)據(jù),沒有數(shù)據(jù)安全

?鏡像集群：是一種主從集群，普通集群的基礎(chǔ)上，添加了主從備份功能，提高集群的數(shù)據(jù)可用性。

鏡像集群雖然支持主從，但主從同步并不是強一致的，某些情況下可能有數(shù)據(jù)丟失的風(fēng)險。因此在RabbitMQ的3.8版本以后，推出了新的功能：仲裁隊列來代替鏡像集群，底層采用Raft協(xié)議確保主從的數(shù)據(jù)一致性。

4.2.普通集群

4.2.1.集群結(jié)構(gòu)和特征

普通集群，或者叫標(biāo)準(zhǔn)集群（classic cluster），具備下列特征：

• 會在集群的各個節(jié)點間共享部分?jǐn)?shù)據(jù)，包括：交換機、隊列元信息。不包含隊列中的消息。數(shù)據(jù)不安全,只有引用關(guān)系
• 當(dāng)訪問集群某節(jié)點時，如果隊列不在該節(jié)點，會從數(shù)據(jù)所在節(jié)點傳遞到當(dāng)前節(jié)點并返回
• 隊列所在節(jié)點宕機，隊列中的消息就會丟失

結(jié)構(gòu)如圖：

4.2.2.docker部署普通rabbitmq集群

因為docker可以環(huán)境隔離,所以采用多個docker 模擬多個機器之間進行連接

1. 在部署之間需要獲取cookie

RabbitMQ底層依賴于Erlang，而Erlang虛擬機就是一個面向分布式的語言，默認(rèn)就支持集群模式。集群模式中的每個RabbitMQ 節(jié)點使用 cookie 來確定它們是否被允許相互通信。

要使兩個節(jié)點能夠通信，它們必須具有相同的共享秘密，稱為Erlang cookie。cookie 只是一串最多 255 個字符的字母數(shù)字字符。

每個集群節(jié)點必須具有相同的 cookie。實例之間也需要它來相互通信。

我們先在之前啟動的mq容器中獲取一個cookie值，作為集群的cookie。執(zhí)行下面的命令：

docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie

可以看到cookie值如下：

接下來，停止并刪除當(dāng)前的mq容器，我們重新搭建集群。

docker rm -f mq

docker 刪除沒有容器使用的數(shù)據(jù)卷

 docker volume prune 

創(chuàng)建文件夾,用于集中管理,一個集群文件夾三個結(jié)點目錄

2.在任一結(jié)點新建配置文件rabbitmq.conf

loopback_users.guest = false
listeners.tcp.default = 5672
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@mq1
cluster_formation.classic_config.nodes.2 = rabbit@mq2
cluster_formation.classic_config.nodes.3 = rabbit@mq3

再創(chuàng)建一個文件，記錄cookie

# 創(chuàng)建cookie文件
touch .erlang.cookie
# 寫入cookie
echo "CHOEKBDJLSGXYLIXNAJL" > .erlang.cookie
# 修改cookie文件的權(quán)限
chmod 600 .erlang.cookie

3. 把每一個配置文件和Cookie復(fù)制到每一個結(jié)點文件夾
   4.創(chuàng)建docker 網(wǎng)絡(luò)

 docker network create  mq-net

5. 運行掛載

 docker run -d --net mq-net -v /home/hadoop/Rabbt-cluster/node1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf/  -v /home/hadoop/Rabbt-cluster/node1/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123 --name mq1 --hostname mq1 -p 8071:5672 -p 8081:15672 rabbitmq:3.8-management

 docker run -d --net mq-net -v /home/hadoop/Rabbt-cluster/node2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf/  -v /home/hadoop/Rabbt-cluster/node2/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123 --name mq2 --hostname mq2 -p 8072:5673 -p 8082:15672 rabbitmq:3.8-management
  docker run -d --net mq-net -v /home/hadoop/Rabbt-cluster/node3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf/  -v /home/hadoop/Rabbt-cluster/node3/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123 --name mq3 --hostname mq3 -p 8073:5674 -p 8083:15672 rabbitmq:3.8-management

6. 查看任一結(jié)點的web端
   就可以看到當(dāng)前結(jié)點列表
   
   任意結(jié)點
   
   創(chuàng)建隊列并且發(fā)消息
   
   其他來個隊列都可以訪問到隊列和消息

4.3.鏡像集群

4.3.1.集群結(jié)構(gòu)和特征

鏡像集群：本質(zhì)是主從模式，具備下面的特征：

• 交換機、隊列、隊列中的消息會在各個mq的鏡像節(jié)點之間同步備份。
• 創(chuàng)建隊列的節(jié)點被稱為該隊列的主節(jié)點，備份到的其它節(jié)點叫做該隊列的鏡像節(jié)點。
• 一個隊列的主節(jié)點可能是另一個隊列的鏡像節(jié)點
• 所有操作都是主節(jié)點完成，然后同步給鏡像節(jié)點
• 主宕機后，鏡像節(jié)點會替代成新的主(主從備份)

結(jié)構(gòu)如圖：

4.3.2.docker部署鏡像隊列模式集群

在剛剛的案例中，一旦創(chuàng)建隊列的主機宕機，隊列就會不可用。不具備高可用能力。如果要解決這個問題，必須使用官方提供的鏡像集群方案。鏡像集群是集群的一種策略配置,所以只需要在剛才搭建的集群集群做指定即可

官方文檔地址：https://www.rabbitmq.com/ha.html

4.1.鏡像模式的特征

默認(rèn)情況下，隊列只保存在創(chuàng)建該隊列的節(jié)點上。而鏡像模式下，創(chuàng)建隊列的節(jié)點被稱為該隊列的主節(jié)點，隊列還會拷貝到集群中的其它節(jié)點，也叫做該隊列的鏡像節(jié)點。

但是，不同隊列可以在集群中的任意節(jié)點上創(chuàng)建，因此不同隊列的主節(jié)點可以不同。甚至，一個隊列的主節(jié)點可能是另一個隊列的鏡像節(jié)點。

用戶發(fā)送給隊列的一切請求，例如發(fā)送消息、消息回執(zhí)默認(rèn)都會在主節(jié)點完成，如果是從節(jié)點接收到請求，也會路由到主節(jié)點去完成。鏡像節(jié)點僅僅起到備份數(shù)據(jù)作用。

當(dāng)主節(jié)點接收到消費者的ACK時，所有鏡像都會刪除節(jié)點中的數(shù)據(jù)。

總結(jié)如下：

• 鏡像隊列結(jié)構(gòu)是一主多從（從就是鏡像）
• 所有操作都是主節(jié)點完成，然后同步給鏡像節(jié)點
• 主宕機后，鏡像節(jié)點會替代成新的主（如果在主從同步完成前，主就已經(jīng)宕機，可能出現(xiàn)數(shù)據(jù)丟失）
• 不具備負載均衡功能，因為所有操作都會有主節(jié)點完成（但是不同隊列，其主節(jié)點可以不同，可以利用這個提高吞吐量）

4.2.鏡像模式的配置

鏡像模式的配置有3種模式來指定鏡像結(jié)點：

這里我們以rabbitmqctl命令作為案例來講解配置語法。
三種模式的優(yōu)缺點
exactly：

• 優(yōu)點：
  這是默認(rèn)的隊列鏡像模式，因此不需要額外的配置。
  消息只會在主節(jié)點和一個從節(jié)點之間進行鏡像，降低了性能開銷。
• 缺點：
  如果主節(jié)點發(fā)生故障，消息只會在一個從節(jié)點上可用，可能會導(dǎo)致消息丟失。
  不提供冗余級別的自動調(diào)整，需要手動配置。

all：

• 優(yōu)點：
  消息在所有從節(jié)點之間進行鏡像，提供了最高級別的冗余和可用性。
  適用于對數(shù)據(jù)持久性和高可用性要求非常高的場景。
• 缺點：
  帶來了較大的性能開銷，因為消息在所有節(jié)點之間同步。
  可能會對硬件資源產(chǎn)生較大的壓力，特別是在大型集群中。

nodes：

• 優(yōu)點：
  這是一種介于exactly和all之間的配置。消息會在指定的一組節(jié)點之間進行鏡像。
  允許您根據(jù)需要選擇特定的從節(jié)點，以平衡性能和冗余級別。
• 缺點：
  需要手動指定節(jié)點，因此需要更多的配置工作。
  需要仔細權(quán)衡冗余和性能需求。

語法示例：

4.2.1.exactly模式

rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

• rabbitmqctl set_policy：固定寫法
• ha-two：策略名稱，自定義
• "^two\."：匹配隊列的正則表達式，符合命名規(guī)則的隊列才生效，這里是任何以two.開頭的隊列名稱
• '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}': 策略內(nèi)容
  • "ha-mode":"exactly"：策略模式，此處是exactly模式，指定副本數(shù)量
  • "ha-params":2：策略參數(shù)，這里是2，就是副本數(shù)量為2，1主1鏡像
  • "ha-sync-mode":"automatic"：同步策略，默認(rèn)是manual，即新加入的鏡像節(jié)點不會同步舊的消息。如果設(shè)置為automatic，則新加入的鏡像節(jié)點會把主節(jié)點中所有消息都同步，會帶來額外的網(wǎng)絡(luò)開銷

進入任意容器進行演示

此時在管理端可以看到策略

只要隊列名字符合正則表達式即可生效
新建隊列

隊列被特殊表示

點+1

就可以看到當(dāng)前主機的鏡像結(jié)點

發(fā)送消息

每一個結(jié)點都可以看到,但是這是普通集群的基本功能消息引用
測試宕機當(dāng)前結(jié)點

此時主節(jié)點宕機mq3,當(dāng)前結(jié)點不會丟失,而是自動尋找健康結(jié)點

重啟后,結(jié)點結(jié)構(gòu)不變

并且之間發(fā)布的消息不會丟失

4.2.2.all模式

rabbitmqctl set_policy ha-all "^all\." '{"ha-mode":"all"}'

• ha-all：策略名稱，自定義
• "^all\."：匹配所有以all.開頭的隊列名
• '{"ha-mode":"all"}'：策略內(nèi)容
  • "ha-mode":"all"：策略模式，此處是all模式，即所有節(jié)點都會稱為鏡像節(jié)點

4.2.3.nodes模式

rabbitmqctl set_policy ha-nodes "^nodes\." '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'

• rabbitmqctl set_policy：固定寫法
• ha-nodes：策略名稱，自定義
• "^nodes\."：匹配隊列的正則表達式，符合命名規(guī)則的隊列才生效，這里是任何以nodes.開頭的隊列名稱
• '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}': 策略內(nèi)容
  • "ha-mode":"nodes"：策略模式，此處是nodes模式
  • "ha-params":["rabbit@mq1", "rabbit@mq2"]：策略參數(shù)，這里指定副本所在節(jié)點名稱

4.4.仲裁隊列

4.4.1.集群特征

仲裁隊列：仲裁隊列是3.8版本以后才有的新功能，用來替代鏡像隊列，具備下列特征：

• 與鏡像隊列一樣，都是主從模式，支持主從數(shù)據(jù)同步
• 使用非常簡單，沒有復(fù)雜的配置
• 主從同步基于Raft協(xié)議，強一致

4.4.2.添加隊列

web端添加,

測試添加仲裁隊列

添加后自動將另外倆個結(jié)點隊列設(shè)置為從節(jié)點,達到主從架構(gòu)

4.5集群動態(tài)擴容

4.5.1.加入集群

1）啟動一個新的MQ容器：

docker run -d --net mq-net \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq4 \
--hostname mq5 \
-p 8074:15672 \
-p 8084:15672 \
rabbitmq:3.8-management

2）進入容器控制臺：

docker exec -it mq4 bash

3）停止mq進程

rabbitmqctl stop_app

4）重置RabbitMQ中的數(shù)據(jù)：

rabbitmqctl reset

5）加入mq1：

rabbitmqctl join_cluster rabbit@mq1

6）再次啟動mq進程

rabbitmqctl start_app

因為仲裁隊列默認(rèn)的鏡像數(shù)為5。如果你的集群有7個節(jié)點，那么鏡像數(shù)肯定是5；而我們集群只有3個節(jié)點，因此鏡像數(shù)量就是3

4.4.3.Java代碼創(chuàng)建仲裁隊列

@Bean
public Queue quorumQueue() {
    return QueueBuilder
        .durable("quorum.queue") // 持久化
        .quorum() // 仲裁隊列
        .build();
}

倆種隊列模式:
仲裁隊列和鏡像隊列有不同的設(shè)計和工作原理，各自適用于不同的使用情境：

• 仲裁隊列：

仲裁隊列使用 Raft 一致性算法來確保數(shù)據(jù)一致性，提供了更強的數(shù)據(jù)冗余和可用性。
仲裁隊列可以在分布式環(huán)境中自動執(zhí)行故障轉(zhuǎn)移，因此在節(jié)點故障時能夠快速恢復(fù)。
適合那些對數(shù)據(jù)一致性和可用性要求非常高的應(yīng)用場景。
鏡像隊列：

• 鏡像隊列

鏡像隊列通過在多個節(jié)點之間復(fù)制消息，提供了數(shù)據(jù)冗余，但不使用 Raft 等一致性算法。
鏡像隊列可以用于提高可用性，但不如仲裁隊列那樣能夠提供強一致性。
可以根據(jù)需要手動配置隊列的鏡像級別。
通常，您可以根據(jù)具體的業(yè)務(wù)需求和性能要求來選擇使用仲裁隊列或鏡像隊列。在某些情況下，特別是對于要求強一致性和自動故障轉(zhuǎn)移的應(yīng)用，仲裁隊列可能是更好的選擇。但在其他情況下，鏡像隊列可能足夠滿足您的需求，并且在性能方面更有效率。

4.4.4.Sp

ringAMQP連接MQ集群

注意，這里用address集群地址來代替host、port方式文章來源地址http://www.zghlxwxcb.cn/news/detail-720751.html

spring:
  rabbitmq:
    addresses: 192.168.150.105:8071, 192.168.150.105:8072, 192.168.150.105:8073
    username: admin
    password: 123
    virtual-host: /

到了這里，關(guān)于Rabbitmq----分布式場景下的應(yīng)用的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！