流量削峰是消息隊(duì)列中常用的場(chǎng)景 一般在秒殺或團(tuán)購(gòu)活動(dòng)中使用廣泛。

使用MQ前：對(duì)于秒殺、搶購(gòu)活動(dòng)，用戶(hù)訪問(wèn)所產(chǎn)生的流量會(huì)很大，甚至?xí)谕粫r(shí)間段出現(xiàn)上萬(wàn)上億條請(qǐng)求，這股瞬間的流量暴漲，我們的應(yīng)用系統(tǒng)配置是無(wú)法承受的，會(huì)導(dǎo)致系統(tǒng)直接崩潰死機(jī)。

例如：A系統(tǒng)平時(shí)每秒請(qǐng)求100個(gè)，系統(tǒng)穩(wěn)定運(yùn)行； 但是晚上8點(diǎn)有秒殺活動(dòng) ，每秒并發(fā)增至1萬(wàn)條 ，系統(tǒng)最大處理每秒1000條 于是導(dǎo)致系統(tǒng)崩潰。?

使用MQ后：我們?cè)诖罅坑脩?hù)進(jìn)行秒殺請(qǐng)求時(shí)，將那個(gè)巨大的流量請(qǐng)求拒在系統(tǒng)業(yè)務(wù)處理的上層，并將其轉(zhuǎn)移至MQ中，而不是直接涌入我們的接口。在這里MQ消息隊(duì)列起到了緩存作用。

例如：100萬(wàn)用戶(hù)在高峰期，每秒請(qǐng)求5000個(gè)，將這5000個(gè)請(qǐng)求寫(xiě)入MQ系統(tǒng)每秒只能處理2000請(qǐng)求，因?yàn)镸ySQL只能處理2000個(gè)請(qǐng)求 ; 系統(tǒng)每秒拉取2000個(gè)請(qǐng)求 不超過(guò)自己的處理能力即可。

使用MQ前：

?使用MQ后：

5.4 其他應(yīng)用場(chǎng)景：

1、訂單處理系統(tǒng)：

在一個(gè)電子商務(wù)平臺(tái)中，可以使用 RabbitMQ 來(lái)處理訂單。當(dāng)用戶(hù)下單時(shí)，訂單信息被發(fā)布到 RabbitMQ 的交換機(jī)中，然后相關(guān)的消費(fèi)者從隊(duì)列中獲取訂單消息并進(jìn)行處理，如驗(yàn)證訂單、庫(kù)存管理、支付等。

2、日志收集與分發(fā)：

假設(shè)有多個(gè)應(yīng)用程序生成日志，并希望將它們集中處理和存儲(chǔ)。每個(gè)應(yīng)用程序可以將日志消息發(fā)布到一個(gè)名為 "log_exchange" 的交換機(jī)中，然后有不同的消費(fèi)者訂閱該交換機(jī)，并將日志消息寫(xiě)入數(shù)據(jù)庫(kù)或發(fā)送到日志分析系統(tǒng)。

3、實(shí)時(shí)數(shù)據(jù)傳輸：

如果有一個(gè)實(shí)時(shí)監(jiān)控系統(tǒng)，需要將傳感器數(shù)據(jù)實(shí)時(shí)傳輸?shù)奖O(jiān)控平臺(tái)進(jìn)行處理和可視化展示。傳感器將數(shù)據(jù)發(fā)布到 RabbitMQ 的交換機(jī)中，監(jiān)控平臺(tái)的消費(fèi)者訂閱交換機(jī)并處理數(shù)據(jù)，從而實(shí)現(xiàn)實(shí)時(shí)監(jiān)控和報(bào)警功能。

4、異步任務(wù)處理：

假設(shè)有一個(gè)應(yīng)用程序需要處理大量耗時(shí)的任務(wù)，如圖像處理、PDF 轉(zhuǎn)換等。應(yīng)用程序?qū)⑷蝿?wù)發(fā)布到 RabbitMQ 的隊(duì)列中，然后有多個(gè)工作節(jié)點(diǎn)作為消費(fèi)者從隊(duì)列中獲取任務(wù)并進(jìn)行處理，以實(shí)現(xiàn)任務(wù)的并行處理和減輕主應(yīng)用程序的壓力。

5、消息通知系統(tǒng)：

?假設(shè)在一個(gè)訂閱系統(tǒng)中，用戶(hù)可以訂閱不同的主題或事件。當(dāng)有新的消息發(fā)布時(shí)，RabbitMQ 會(huì)將消息路由到對(duì)應(yīng)的隊(duì)列，然后訂閱該隊(duì)列的用戶(hù)會(huì)收到相應(yīng)的通知。這種方式可以用于實(shí)現(xiàn)郵件訂閱、新聞推送等功能。

6、微服務(wù)架構(gòu)：

在一個(gè)微服務(wù)架構(gòu)中，不同的服務(wù)之間可能需要進(jìn)行消息傳遞和協(xié)作。使用 RabbitMQ 可以實(shí)現(xiàn)服務(wù)之間的解耦和異步通信，每個(gè)服務(wù)通過(guò)交換機(jī)和隊(duì)列收發(fā)消息，從而實(shí)現(xiàn)微服務(wù)之間的松耦合。

六、?RabbitMQ 安裝

官網(wǎng)地址：

RabbitMQ: easy to use, flexible messaging and streaming — RabbitMQ

6.1 brew安裝

brew update #更新一下homebrew
brew install rabbitmq ? #安裝rabbitMQ?

安裝結(jié)果：

==> Caveats
==> rabbitmq
Management Plugin enabled by default at http://localhost:15672

To restart rabbitmq after an upgrade:
? brew services restart rabbitmq
Or, if you don't want/need a background service you can just run:
? CONF_ENV_FILE="/opt/homebrew/etc/rabbitmq/rabbitmq-env.conf" /opt/homebrew/opt/rabbitmq/sbin/rabbitmq-server

?rabbitmq 的安裝路徑：

/opt/homebrew/opt/rabbitmq

6.2 、配置環(huán)境變量

1、

vi ~/.bash_profile

2、?

export RABBIT_HOME=${PATH}:/opt/homebrew/opt/rabbitmq
export PATH=${PATH}:$RABBIT_HOME/sbin

3、?

source ~/.bash_profile

6.3 、啟動(dòng)RabbitMQ

1、前臺(tái)運(yùn)行
rabbitmq-server

2、后臺(tái)運(yùn)行
rabbitmq-server -detached

3、查看運(yùn)行狀態(tài)
rabbitmqctl status

4、開(kāi)始 Web插件
rabbitmq-plugins enable rabbitmq_management

5、重啟
rabbitmq-server restart

5、關(guān)閉
rabbitmqctl stop

6.4、訪問(wèn)MQ

1、瀏覽器地址

http://localhost:15672/
默認(rèn)用戶(hù)名和密碼為guest

添加用戶(hù)
rabbitmqctl add_user miaojiang 123

設(shè)置用戶(hù)為管理員
rabbitmqctl set_user_tags miaojiang administrator

配置用戶(hù)可以遠(yuǎn)程登錄
rabbitmqctl set_permissions -p "/" miaojaing ".*" ".*" ".*"

查看新添加的賬戶(hù)

rabbitmqctl list_users

查看用于的權(quán)限
rabbitmqctl list_permissions -p /

七、Spring Boot 項(xiàng)目應(yīng)用RabbitMQ

7.1、添加Maven依賴(lài)：

在你的項(xiàng)目的pom.xml文件中添加RabbitMQ客戶(hù)端庫(kù)的依賴(lài)

<!--AMQP依賴(lài)，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

7.2、配置RabbitMQ連接：

在Spring Boot的配置文件(application.properties 或 application.yml)中添加RabbitMQ的連接信息。

application.properties:

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

在application.yml配置mq的參數(shù)：

spring:
  rabbitmq:
    #設(shè)置RabbitMQ的IP地址
    host: localhost
    #設(shè)置rabbitmq服務(wù)器用戶(hù)名
    username: guest
    #設(shè)置rabbitmq服務(wù)器密碼
    password: guest
    #設(shè)置rabbitmq服務(wù)器連接端口
    port: 5672

7.3?創(chuàng)建交換機(jī)

自定義交換機(jī)名稱(chēng)

創(chuàng)建名為“myExchange”的交換機(jī)

package com.example.usermanagement.mq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    /*
    使用 @Configuration 注解創(chuàng)建一個(gè)配置類(lèi)，并通過(guò) @Bean 注解創(chuàng)建了一個(gè)名為 declareExchange 的方法，用于聲明創(chuàng)建交換機(jī)。請(qǐng)根據(jù)實(shí)際情況修改交換機(jī)名稱(chēng)、類(lèi)型和持久化設(shè)置。
     */

    public static final String EXCHANGE_NAME = "myExchange";

    @Bean
    public Exchange declareExchange() {
        return ExchangeBuilder.directExchange(EXCHANGE_NAME)
                .durable(true)
                .build();
    }
}

7.4?創(chuàng)建消息發(fā)送者

創(chuàng)建消息發(fā)送者：創(chuàng)建一個(gè)消息發(fā)送者的類(lèi)，用于發(fā)送消息到RabbitMQ

package com.example.usermanagement.mq;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MessageSender{

    private final AmqpTemplate amqpTemplate;
    private final String exchangeName = "myExchange"; // 自定義交換機(jī)名稱(chēng)

    @Autowired
    public MessageSender(AmqpTemplate amqpTemplate) {
        this.amqpTemplate = amqpTemplate;
    }

    public void sendMessage(Object message) {
        amqpTemplate.convertAndSend(exchangeName, "", message); // 發(fā)送消息到默認(rèn)交換機(jī)和空路由鍵
    }
}

注意：

?sendMessage 類(lèi)型使用的是Object類(lèi)型

7.5?RabbitMQ管理后臺(tái)添加對(duì)列

步驟：

1. 打開(kāi)瀏覽器，輸入RabbitMQ管理后臺(tái)的URL。默認(rèn)情況下，該URL為http://localhost:15672/。請(qǐng)確保你的RabbitMQ服務(wù)器正在運(yùn)行，并且端口號(hào)正確。

2. 輸入用戶(hù)名和密碼以登錄到RabbitMQ管理后臺(tái)。默認(rèn)情況下，用戶(hù)名為guest，密碼也為guest。如果你修改過(guò)用戶(hù)名和密碼，請(qǐng)使用你的自定義憑據(jù)進(jìn)行登錄。

3. 成功登錄后，你將看到RabbitMQ管理后臺(tái)的主界面。在頂部導(dǎo)航欄中，選擇Queues選項(xiàng)卡。

4. 在Queues頁(yè)面上，你將看到已經(jīng)存在的隊(duì)列列表。如果你想要?jiǎng)?chuàng)建一個(gè)新隊(duì)列，請(qǐng)點(diǎn)擊Add a new queue按鈕。

5. 在添加隊(duì)列的頁(yè)面上，填寫(xiě)以下信息：

   • Name：隊(duì)列的名稱(chēng)。為隊(duì)列提供一個(gè)唯一的名稱(chēng)。（如myQueue)
   • Durability：隊(duì)列的持久性。選擇是或否，以指定隊(duì)列是否應(yīng)該在RabbitMQ服務(wù)重啟后保留。
   • Auto delete：隊(duì)列的自動(dòng)刪除。選擇是或否，以指定當(dāng)最后一個(gè)消費(fèi)者斷開(kāi)連接后，是否刪除隊(duì)列。
   • Arguments：隊(duì)列的其他參數(shù)。這是可選的，你可以為隊(duì)列設(shè)置一些特定的參數(shù)。

6. 填寫(xiě)完隊(duì)列信息后，點(diǎn)擊Add queue按鈕以創(chuàng)建隊(duì)列。

7. 創(chuàng)建成功后，你將在Queues頁(yè)面上看到新添加的隊(duì)列。你可以在該頁(yè)面上查看隊(duì)列的詳細(xì)信息，包括消息數(shù)量、消費(fèi)者數(shù)量等。

http://localhost:15672/#/queues

只需要添加隊(duì)列名稱(chēng)就可以?

7.6?調(diào)用生產(chǎn)者

1、注入MessageSender實(shí)例

@Autowired
private MessageSender messageSender;

2、在需要發(fā)送消息的地方調(diào)用messageSender.sendMessage方法。根據(jù)你的業(yè)務(wù)邏輯，你可以在合適的位置調(diào)用該方法。例如，在訂單創(chuàng)建成功后，你可以添加以下代碼：

messageSender.sendMessage("訂單已創(chuàng)建：" + order.getOrderId());

7.7?創(chuàng)建消息接收者

創(chuàng)建消息接收者：創(chuàng)建一個(gè)消息接收者的類(lèi)，用于處理接收到的RabbitMQ消息。

這里就直接寫(xiě)處理RabbitMQ消息的邏輯。

package com.example.usermanagement.mq;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MessageReceiver {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("your_queue_name"),
            exchange = @Exchange(value = RabbitMQConfig.EXCHANGE_NAME)
//            key = "your_routing_key"
    ))
    public void receiveMessage(Object message) {
        System.out.println("Received message: " + message);
        // 處理消息邏輯
    }
}

注意：

?sendMessage 類(lèi)型使用的是Object類(lèi)型

your_queue_name 替換為你要監(jiān)聽(tīng)的隊(duì)列的名稱(chēng),（如myQueue)

將 your_routing_key 替換為適當(dāng)?shù)穆酚涉I（如果使用）

八、MQ界面介紹

Overview（概覽）：提供了一份概覽報(bào)告，包括服務(wù)器和集群信息、節(jié)點(diǎn)狀態(tài)、隊(duì)列和連接摘要、以及最近的相關(guān)日志條目。
Connections（連接）：顯示當(dāng)前連接到 RabbitMQ 服務(wù)器的客戶(hù)端應(yīng)用程序，包括連接的名稱(chēng)、協(xié)議、虛擬主機(jī)等信息。
Channels（通道）：顯示每個(gè)連接上的活動(dòng)通道，以及與每個(gè)通道相關(guān)的一些指標(biāo)，如消費(fèi)者數(shù)量、未確認(rèn)的消息數(shù)量等。
Exchanges（交換機(jī)）：列出了所有的交換機(jī)，包括名稱(chēng)、類(lèi)型、綁定的隊(duì)列和綁定的數(shù)量。
Queues（隊(duì)列）：顯示了所有的隊(duì)列，包括名稱(chēng)、消息數(shù)量、消費(fèi)者數(shù)量等信息。您還可以通過(guò)隊(duì)列進(jìn)行一些操作，如創(chuàng)建、刪除、清空等。
Admin（管理員）：提供了一些高級(jí)管理功能，如用戶(hù)和權(quán)限管理、虛擬主機(jī)管理、插件管理等。

8.1?Overview（概覽）

Overview（概覽）：提供了一份概覽報(bào)告，包括服務(wù)器和集群信息、節(jié)點(diǎn)狀態(tài)、隊(duì)列和連接摘要、以及最近的相關(guān)日志條目。

?

8.2?Connections（連接）

Connections（連接）：顯示當(dāng)前連接到 RabbitMQ 服務(wù)器的客戶(hù)端應(yīng)用程序，包括連接的名稱(chēng)、協(xié)議、虛擬主機(jī)等信息。

8.3?Channels（通道）

Channels（通道）：顯示每個(gè)連接上的活動(dòng)通道，以及與每個(gè)通道相關(guān)的一些指標(biāo)，如消費(fèi)者數(shù)量、未確認(rèn)的消息數(shù)量等。

8.3.1?prefetch（預(yù)?。?/h4>

在消息隊(duì)列（Message Queue）中，prefetch（預(yù)?。┦且粋€(gè)重要的概念，它用于控制消費(fèi)者從消息隊(duì)列中獲取消息的速度。

Prefetch 是指在消費(fèi)者端從消息隊(duì)列中獲取消息之前，先獲取一定數(shù)量的消息到本地緩存中，以供消費(fèi)者快速處理。這樣可以提高系統(tǒng)的吞吐量和效率，減少網(wǎng)絡(luò)傳輸?shù)拈_(kāi)銷(xiāo)。

具體來(lái)說(shuō)，prefetch 可以幫助避免以下情況發(fā)生：

1. 公平分發(fā)（Fair dispatching）：在多個(gè)消費(fèi)者并行處理消息時(shí)，如果沒(méi)有 prefetch 限制，一個(gè)消費(fèi)者可能會(huì)一次性獲取到過(guò)多的消息，導(dǎo)致其他消費(fèi)者得到較少的機(jī)會(huì)。通過(guò)設(shè)置 prefetch，可以確保每個(gè)消費(fèi)者只能獲取一定數(shù)量的消息，實(shí)現(xiàn)更公平的消息分發(fā)。

2. 消費(fèi)者負(fù)載均衡（Consumer load balancing）：當(dāng)消息隊(duì)列中有大量待處理的消息時(shí)，消費(fèi)者可能會(huì)因?yàn)樘幚硭俣容^慢而積壓消息。通過(guò)設(shè)置合適的 prefetch 值，可以限制消費(fèi)者每次獲取的消息數(shù)量，使得消息能夠均勻地分配給多個(gè)消費(fèi)者，從而實(shí)現(xiàn)負(fù)載均衡。

在 RabbitMQ 中，prefetch 的設(shè)置可以通過(guò) basic.qos 方法來(lái)進(jìn)行配置。例如，以下代碼將設(shè)置 prefetch 數(shù)量為 10：

channel.basic_qos(prefetch_count=10)

請(qǐng)注意，prefetch 的設(shè)置應(yīng)該根據(jù)具體的應(yīng)用場(chǎng)景和系統(tǒng)負(fù)載情況進(jìn)行調(diào)優(yōu)。合理設(shè)置 prefetch 可以提高系統(tǒng)的性能和穩(wěn)定性。

8.4 Exchanges（交換機(jī)）

Exchanges（交換機(jī)）：列出了所有的交換機(jī)，包括名稱(chēng)、類(lèi)型、綁定的隊(duì)列和綁定的數(shù)量。

8.5 Queues（隊(duì)列）

Queues（隊(duì)列）：顯示了所有的隊(duì)列，包括名稱(chēng)、消息數(shù)量、消費(fèi)者數(shù)量等信息。您還可以通過(guò)隊(duì)列進(jìn)行一些操作，如創(chuàng)建、刪除、清空等。

8.6?Admin（管理員）

Admin（管理員）：提供了一些高級(jí)管理功能，如用戶(hù)和權(quán)限管理、虛擬主機(jī)管理、插件管理等文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-656857.html

到了這里，關(guān)于RabbitMQ：可靠消息傳遞的強(qiáng)大消息中間件的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！