RabbitMQ中間件

来源:https://www.cnblogs.com/maomao777/archive/2023/02/22/rabbitMQ_knowledge.html
-Advertisement-
Play Games

RabbitMQ 配置環境 安裝 erlang環境以及RabbitMQ RabbitMQ埠號： 5672 去官網下載 https://www.rabbitmq.com 然後重啟RabbitMQ服務 RabbitMQ安裝教程 開放埠15672 這裡，通過http://IP地址:15672 進行We ...


RabbitMQ

配置環境

安裝 erlang環境以及RabbitMQ

RabbitMQ埠號： 5672

去官網下載 https://www.rabbitmq.com

然後重啟RabbitMQ服務 RabbitMQ安裝教程

開放埠15672

這裡，通過http://IP地址:15672 進行Web頁面登錄，輸入賬號密碼（預設都是guest），完成頁面訪問。至此，全部安裝結束。

導入依賴

<!-- 集成RabbitMQ -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置相關信息

RabbitMQ的埠號是什麼？

5672 ：這是rabbitMQ的埠號；
15672 ：這是那個RabbitMQ的web頁面的埠號；

spring.application.name=spirng-boot-rabbitmq
spring.rabbitmq.host=127.0.0.1 ##主機ip
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=MmHost
spring.rabbitmq.username=root
spring.rabbitmq.password=root

spring:
    rabbitmq:
        username: root
        password: root
        addresses: 127.0.0.1:5672
        cache:
        connection:
        #Cache connection mode, with default connections and multiple channels
        mode: channel
        #Multiple connections, multiple channels
        # mode: connection

# rabbitmq
server:
  port: 8080
spring:
  #給項目來個名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 伺服器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虛擬host 可以不設置,使用server預設host
    virtual-host: MmHost

發送消息

@Component
public class SenderTest{
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void Send() {
        // 隊列名稱
        String queueName = "ThisKey";
        // 消息
        String message = "Hello, Spring AMQP!";
        // 發送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

rabbitTemplate.convertAndSend(key, message);

接收消息

@Component
public class SpringRabbitMQListener {
    @RabbitListener(queues = "ThisKey")
    public void listenSimpleQueueMsg(String msg){
        System.out.println(msg);
    }
}

@Component
//指定所監聽的隊列
@RabbitListener(queues = "ThisKey")
public class SpringRabbitMQListener {
    //指定用來處理接收消息的方法
    @RabbitHandler
    public void listenSimpleQueueMsg(String msg){
        System.out.println(msg);
    }
}

註意：此處消息被消費後，對應的ThisKey中的消息就消失了。

原文鏈接 去的去看看

RabbitMQ-基礎使用(Spring AMQP) - 簡書 (jianshu.com)

如果使用其他交換機，則需要進行相關配置

可以看這篇文章：SpringBoot整合RabbitMQ

1、創建對應的配置文件

例如Direct交換機

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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class DirectRabbitConfig {
 
    //隊列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue",true);
    }
 
    //Direct交換機 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange");
    }
 
    //綁定  將隊列和交換機綁定, 並設置用於匹配鍵：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
}

如何保證消息的可靠？

ack應答

消息應答

概念
消費者完成一個任務可能需要一段時間，如果其中一個消費者處理一個長的任務並僅只完成了部分突然它掛掉了，會導致消息丟失。RabbitMQ 一旦向消費者傳遞了一條消息，便立即將該消息標記為刪除。在這種情況下，突然有個消費者掛掉了，我們將丟失正在處理的消息。以及後續發送給該消費這的消息，因為它無法接收到。

為了保證消息在發送過程中不丟失，rabbitmq 引入消息應答機制，消息應答就是:消費者在接收到消息並且處理該消息之後，告訴 rabbitmq 它已經處理了，rabbitmq 可以把該消息刪除了。

各種消息模型實例

五種交換機類型

Direct Exchange

直連型交換機，根據消息攜帶的路由鍵將消息投遞給對應隊列。

大致流程，有一個隊列綁定到一個直連交換機上，同時賦予一個路由鍵 routing key 。
然後當一個消息攜帶著路由值為X，這個消息通過生產者發送給交換機時，交換機就會根據這個路由值X去尋找綁定值也是X的隊列。

Fanout Exchange

扇型交換機，這個交換機沒有路由鍵概念，就算你綁了路由鍵也是無視的。 這個交換機在接收到消息後，會直接轉發到綁定到它上面的所有隊列。

Topic Exchange

主題交換機，這個交換機其實跟直連交換機流程差不多，但是它的特點就是在它的路由鍵和綁定鍵之間是有規則的。

其他兩種：WorkQueue（通過監聽隊列名稱）

基本消息隊列BasicQueue即為上方的代碼，此處不再重覆。

- 1、WorkQueue（通過監聽隊列名稱）

img

WorkQueue.png

WorkQueue與BasicQueue不同之處，就是WorkQueue支持一對多發佈消息（不是一個消息發給多個消費者，一個消息只會被一個消費者消費），多個消費者可以提高消息消費速度，當然相同之處也是消息消費後就會從Queue中消失（後續的幾種模型都是如此）。

① 模擬消息堆積

    // 隊列名稱
    String queueName = "simple.queue";
    // 消息
    String message = "Message_";
    for (int i = 1; i <= 50; i++) {
        // 發送消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }

② 接收消息
此處設置兩個線程處理速度不同。

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消費者1接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消費者2........接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}

處理結果是2個消費者會均分消息。可以修改消費方的配置，以按照實際處理能力分配，如下：

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能獲取一條消息，處理完成才能獲取下一個消息

- 2、Fanout （扇形交換機）

img

Fanout.png

① 編寫Fanout配置類
創建FanoutExchange，綁定隊列Queue和交換機Exchange。

@Configuration
public class FanoutConfig {
    /**
     * 聲明交換機
     * @return Fanout類型交換機
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("stone.fanout");
    }
    /**
     * 第1個隊列
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    /**
     * 綁定隊列和交換機
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    /**
     * 第2個隊列
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    /**
     * 綁定隊列和交換機
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

② 發送消息

    // 隊列名稱
    String exchangeName = "stone.fanout";
    // 消息
    String message = "Hello, Fanout!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);

③ 接收消息

    @RabbitListener(queues = "fanout.queue1")
    public void listen1FanoutQueueMsg(String msg){
        System.out.println("Listener1 get :" + msg);
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listen2FanoutQueueMsg(String msg){
        System.out.println("Listener2 get :" + msg);
    }

不同於WorkQueue，Fanout Exchange廣播模型下，綁定該交換機的消費者可以獲取到對應的消息（即一條消息可以通過交換機被多個消費者消費）。

- 3、Direct（直連交換機）

img

Direct.png

① 基於註解聲明隊列和交換機
@RabbitListener的使用
Ⅰ bindings = @QueueBinding()配置綁定關係；
Ⅱ value = @Queue(name = "direct.queue1")配置隊列；
Ⅲ exchange = @Exchange(name = "stone.direct", type = ExchangeTypes.DIRECT)配置交換機；
Ⅳ key = {"talkshow", "musicshow"}配置訂閱。監聽的key進行匹配，
rabbitTemplate.convertAndSend(exchangeName, "xxx", message);中的xxx第二個參數進行匹配
註意：type = ExchangeTypes.DIRECT是預設類型，可以不做配置。

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "stone.direct", type = ExchangeTypes.DIRECT),
            key = {"talkshow", "musicshow"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("DirectQueue1 :" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "stone.direct", type = ExchangeTypes.DIRECT),
            key = {"talkshow", "news"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("DirectQueue2 :" + msg);
    }

② 發送消息

    // 交換機名稱
    String exchangeName = "itcast.direct";
    // 消息
    String messageNews = "烏俄衝突升級，昔日友邦冷眼旁觀！";
    // 發送消息
    rabbitTemplate.convertAndSend(exchangeName, "news", messageNews);
    // 消息
    String messageTalks = "蜘蛛俠3英雄無歸發佈藍光預告，主演再登SN宣傳！";
    // 發送消息
    rabbitTemplate.convertAndSend(exchangeName, "talkshow", messageTalks);

此時：訂閱news主題的隊列direct.queue1可以消費messageNews，訂閱talkshow主題的direct.queue1和direct.queue2均可以消費messageTalks。

- 4、Topic（主題交換機）

img

Topic.png

Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。只不過Topic類型Exchange可以讓隊列在綁定Routing key 的時候使用通配符。通配符規則：
#：匹配一個或多個詞
*：匹配不多不少恰好1個詞
① 發送消息

/**
     * topicExchange
     */
@Test
public void testSendTopicExchange() {
    // 交換機名稱
    String exchangeName = "itcast.topic";
    // 消息
    String message = "建設更高水平法治中國";
    // 發送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

② 接收消息

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "stone.topic", type = ExchangeTypes.TOPIC),
            key = {"China.#"}
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("TopicQueue1 :" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "stone.topic", type = ExchangeTypes.TOPIC),
            key = {"#.news"}
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("TopicQueue2 :" + msg);
    }

此時，由於消息Topic滿足兩個隊列的訂閱規則，所以兩個隊列都可以消費到消息。

RabbitMQ的應用場景

  1. 延遲隊列，延遲消息
  2. 服務與服務之間的解耦（例如一個服務進行mysql操作的時候需要另一個服務同時進行對應操作）
  3. 非同步處理、流量削峰

1、非同步處理

假設想象一下我們做一個商城項目，在用戶支付模塊中，可能會涉及到其它業務，比如：積分折扣、消費券、簡訊驗證等功能。我們傳統的執行步驟是逐步執行，也就是說當用戶點擊支付 ----> 積分折扣 ----> 消費券 ----> 簡訊驗證 ----->支付完成，用戶需要等待每個業務執行完畢才能支付成功！假設我們從點擊支付 -----> 支付成功消耗時間為100/ms，後面我們每新增一個業務就會多耗時50/ms，上述的流程大概會耗時250/ms！如果說以後業務更多的話，那麼用戶支付訂單的時間會越來越長，這樣大大影響了用戶的體驗！參照下圖理解
非同步處理演示圖

我們使用消息中間件進行非同步處理，當用戶下單支付同時我們創建消息隊列進行非同步的處理其它業務，在我們支付模塊中最重要的是用戶支付，我們可以將一些不重要的業務放入消息隊列執行，這樣可以大大添加我們程式運行的速度，用戶支付模塊中也大大減少了支付時間，為用戶添加了更好的體驗。其它模塊與其思想一致，就比如說用戶註冊！

2、流量削峰

假設我們有一個訂單系統，我們的訂單系統最大承受訪問量是每秒1萬次，如果說某天訪問量過大我們的系統承受不住了，會對伺服器造成宕機，這樣的話我們的系統就癱瘓了，為瞭解決該問題我們可以使用中間件對流量進行消峰

未加入中間件之前，用戶直接訪問的是訂單系統
未加入中間件

加入中間件之後，用戶直接訪問的是中間件，通過中間件對用戶進行消峰，好處是可以避免系統的宕機癱瘓，壞處是系統速度變慢，但是總比不能使用好
加入中間件

3、應用解耦

我們以商城項目為例，訂單系統耦合調用支付、庫存、物流系統，如果某天其中一個系統出現了異常就會造成訂單系統故障！使用中間件後訂單系統通過隊列去訪問支付、庫存、物流系統就不會造成上述的問題，因為訂單系統執行完成才會發消息給隊列，接下來的任務就交給隊列完成，隊列會監督各個系統完成，如果完不成隊列會一直監督，直到完成為止！所以說使用中間件後不會造成一個子系統出現故障而造成整個系統故障
應用解耦

本文來自博客園，作者：沒有煩惱的貓貓，轉載請註明原文鏈接：https://www.cnblogs.com/maomao777/p/rabbitMQ_knowledge.html


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 使用VS code將本地的項目推送到gitee的方法（圖例）
    第一步：在gitee新建一個倉庫 寫上基本信息，點擊創建（不要勾選初始化），記住項目名稱 成功後的頁面中，把這個項目地址複製好 第二步VS code推送至gitee 把項目地址粘貼在輸入框，回車 再輸入剛纔新建的項目名稱，回車 此時第一次提交會有新彈窗界面，提示輸入gitee的用戶名和密碼，輸入後點 ...
  • JavaScript 中字元串和數組的截取辦法總結
    字元串 substring 形式：substring(index1, index2) 不會改變原始字元串 將小的參數作為開始位置 start，大的作為結束位置 stop 包含 start，但不包含 stop 如果只給一個參數表示從它 start 截取到字元串結尾 如果某個參數為負，會先將負數變為 0 ...
  • 數值的擴展方法以及新增數據類型BigInt
    二進位和八進位表示法 ES6提供了二進位和八進位數值的新的寫法，分別用首碼0b(或0B)和0o或(0O)表示 0b111110111 503 // true; 0o767 503; // true 在es5開始，嚴格模式中，八進位不允許使用首碼0 表示，否則會報錯 // 嚴格模式 (function ...
  • 前端頁面開發less規範寫法
    代碼組織 代碼按一下順序組織： @import 變數聲明 樣式聲明 1 @import "mixins/size.less"; 2 @default-text-color: #333; 3 .page { 4 width: 960px; 5 margin: 0 auto; 6 } @import 語 ...
  • 回顧過去展望未來，移動開發前景分析
    vue2 使用 cesium 篇 今天好好寫一篇哈，之前寫的半死不活的。首先說明：這篇博文是我邊做邊寫的，小白也是，實現效果會同時發佈截圖，如果沒有實現也會說明，僅僅作為技術積累，選擇性分享，不做教學哈。不好別噴。 安裝 cesium 這個就很簡單，只需要一句簡簡單單的命令就可以實現在 vue 項目 ...
  • 現代圖片性能優化及體驗優化指南 - 響應式圖片方案
    本文是系列第二篇。系列文章： 現代圖片性能優化及體驗優化指南 - 圖片類型及 Picture 標簽的使用 圖片資源，在我們的業務中可謂是占據了非常大頭的一環，尤其是其對帶寬的消耗是十分巨大的。 對圖片的性能優化及體驗優化在今天就顯得尤為重要。本文，就將從各個方面闡述，在各種新特性滿頭飛的今天，我們可 ...
  • 我用chatGPT做直播技術選型，捲死了同事
    一、背景 遠程服務將電腦程式的工作範圍從單機擴展到網路，從本地延伸至遠程，是構建分散式系統的首要基礎。遠程服務調用（Remote Procedure Call，RPC）在電腦科學中已經存在了超過四十年時間。但很多人無法明確區分RPC與Rest。本文就講一講RPC和Rest的本質區別。 二、分析 ...
  • （原創）多線程併發：AQS源碼分析（1）——獨占鎖的實現原理
    談到java中的併發，我們就避不開線程之間的同步和協作問題，談到線程同步和協作我們就不能不談談jdk中提供的AbstractQueuedSynchronizer(翻譯過來就是抽象的隊列同步器)機制； (一)、AQS中的state和Node含義： AQS中提供了一個int volatile state ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類