【Spring開發】SpringCloud服務端基礎框架第4篇：3.2.WorkQueue,3.3.發佈/訂閲【附代碼文详情 - spring,java,ssh,後端小帥説技術日志

动态

详情

【Spring開發】SpringCloud服務端基礎框架第4篇：3.2.WorkQueue,3.3.發佈/訂閲【附代碼文 - 动态详情

02:17 下午 · 10月 26 ,2025

🏆🏆🏆教程全知識點簡介：Docker實用篇 0.學習目標 1.初識Docker 1.2.Docker和虛擬機的區別 2.Docker的基本操作 2.1.鏡像操作 2.1.3.案例1-拉取、查看鏡像 2.1.5.練習 3.Dockerfile自定義鏡像 3.3.構建Java項目 3.4.小結 4.Docker-Compose 4.3.部署微服務集羣 4.3.1.compose文件 4.3.3.打包設置docker鏡像源關閉禁止開機啓動防火牆安裝修改權限補全命令 RabbitMQ 1.初識MQ 1.1.同步和異步通訊 1.2.技術對比： 2.快速入門 2.2.RabbitMQ消息模型 2.3.導入Demo工程 2.4.入門案例 2.4.1.publisher實現 3.1.3.測試 3.2.WorkQueue 3.2.1.消息發送 3.2.4.能者多勞 3.3.發佈/訂閲 3.4.Fanout 3.4.1.聲明隊列和交換機 3.5.Direct 3.6.Topic 3.6.1.説明 SpringCloud01 1.認識微服務 1.1.單體架構 2.服務拆分和遠程調用 2.1.服務拆分原則 2.3.實現遠程調用案例 2.3.1.案例需求： 2.3.2.註冊RestTemplate 3）啓動多個user-service實 4.Ribbon負載均衡 4.1.負載均衡原理 4.2.源碼跟蹤 1）LoadBalancerIntercepor 3）負載均衡策略IRule 5.Nacos註冊中心 5.1.認識和安裝Nacos 5.2.服務註冊到nacos 1）引入依賴 2）配置nacos地址 3）重啓 5.3.服務分級存儲模型 5.3.2.同集羣優先的負載均衡 5.4.權重配置 Nacos安裝指南 1.Windows安裝 1.1.下載安裝包 1.2.解壓 1.3.端口配置 1.4.啓動 1.5.訪問 2.Linux安裝 2.1.安裝JDK 3.Gateway服務網關 3.1.為什麼需要網關 3.3.斷言工廠

📚📚👉👉👉git倉庫code.zip 直接get： https://gitlab.com/yiqing112/backend/-/blob/main/Spring/Sprin... 🍅🍅

✨ 本教程項目亮點

🧠 知識體系完整：覆蓋從基礎原理、核心方法到高階應用的全流程內容
💻 全技術鏈覆蓋：完整前後端技術棧，涵蓋開發必備技能
🚀 從零到實戰：適合 0 基礎入門到提升，循序漸進掌握核心能力
📚 豐富文檔與代碼示例：涵蓋多種場景，可運行、可複用
🛠 工作與學習雙參考：不僅適合系統化學習，更可作為日常開發中的查閲手冊
🧩 模塊化知識結構：按知識點分章節，便於快速定位和複習
📈 長期可用的技術積累：不止一次學習，而是能伴隨工作與項目長期參考

🎯🎯🎯全教程總章節

🚀🚀🚀本篇主要內容

3.2.WorkQueue

Work queues，也被稱為（Task queues），任務模型。簡單來説就是讓多個消費者綁定到一個隊列，共同消費隊列中的消息。

當消息處理比較耗時的時候，可能生產消息的速度會遠遠大於消息的消費速度。長此以往，消息就會堆積越來越多，無法及時處理。

此時就可以使用work 模型，多個消費者共同處理消息處理，速度就能大大提高了。

3.2.1.消息發送

這次循環發送，模擬大量消息堆積現象。

在publisher服務中的SpringAmqpTest類中添加一個測試方法：

/**
     * workQueue
     * 向隊列中不停發送消息，模擬消息堆積。
     */
@Test
public void testWorkQueue() throws InterruptedException {
    // 隊列名稱
    String queueName = "simple.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) {
        // 發送消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

3.2.2.消息接收

要模擬多個消費者綁定同一個隊列，在consumer服務的SpringRabbitListener中添加2個新的方法：

@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);
}

注意到這個消費者sleep了1000秒，模擬任務耗時。

3.2.3.測試

啓動ConsumerApplication後，在執行publisher服務中剛剛編寫的發送測試方法testWorkQueue。

可以看到消費者1很快完成了自己的25條消息。消費者2卻在緩慢的處理自己的25條消息。

也就是説消息是平均分配給每個消費者，並沒有考慮到消費者的處理能力。這樣顯然是有問題的。

NetBeans 文檔

3.2.4.能者多勞

在spring中有一個簡單的配置，可以解決這個問題。修改consumer服務的application.yml文件，添加配置：

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

3.2.5.總結

Work模型的使用：

多個消費者綁定到一個隊列，同一條消息只會被一個消費者處理
通過設置prefetch來控制消費者預取的消息數量

3.3.發佈/訂閲

發佈訂閲的模型如圖：

可以看到，在訂閲模型中，多了一個exchange角色，而且過程略有變化：

Publisher：生產者，也就是要發送消息的程序，但是不再發送到隊列中，而是發給X（交換機）
Exchange：交換機，圖中的X。一方面，接收生產者發送的消息。另一方面，知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決於Exchange的類型。Exchange有以下3種類型：
- Fanout：廣播，將消息交給所有綁定到交換機的隊列
- Direct：定向，把消息交給符合指定routing key 的隊列
- Topic：通配符，把消息交給符合routing pattern（路由模式）的隊列
Consumer：消費者，與以前一樣，訂閲隊列，沒有變化
Queue：消息隊列也與以前一樣，接收消息、緩存消息。

Exchange（交換機）只負責轉發消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那麼消息會丟失！

3.4.Fanout

Fanout，英文翻譯是扇出，我覺得在MQ中叫廣播更合適。

在廣播模式下，消息發送流程是這樣的：

1）可以有多個隊列

Spring Cloud 文檔

2）每個隊列都要綁定到Exchange（交換機）
3）生產者發送的消息，只能發送到交換機，交換機來決定要發給哪個隊列，生產者無法決定
4）交換機把消息發送給綁定過的所有隊列
5）訂閲隊列的消費者都能拿到消息

的計劃是這樣的：

創建一個交換機 itcast.fanout，類型是Fanout
創建兩個隊列fanout.queue1和fanout.queue2，綁定到交換機itcast.fanout

3.4.1.聲明隊列和交換機

Spring提供了一個接口Exchange，來表示所有不同類型的交換機：

在consumer中創建一個類，聲明隊列和交換機：

Spring Boot 文檔

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    /**
     * 聲明交換機
     * @return Fanout類型交換機
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.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);
    }
}

3.4.2.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

@Test
public void testFanoutExchange() {
    // 隊列名稱
    String exchangeName = "itcast.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

3.4.3.消息接收

在consumer服務的SpringRabbitListener中添加兩個方法，作為消費者：

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
    System.out.println("消費者1接收到Fanout消息：【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
    System.out.println("消費者2接收到Fanout消息：【" + msg + "】");
}

Druid 連接池

3.4.4.總結

交換機的作用是什麼？

接收publisher發送的消息
將消息按照規則路由到與之綁定的隊列
不能緩存消息，路由失敗，消息丟失
FanoutExchange的會將消息路由到每個綁定的隊列

聲明隊列、交換機、綁定關係的Bean是什麼？

Queue
FanoutExchange
Binding

3.5.Direct

在Fanout模式中，一條消息，會被所有訂閲的隊列都消費。但是，在某些場景下，希望不同的消息被不同的隊列消費。這時就要用到Direct類型的Exchange。

在Direct模型下：

隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）
消息的發送方在向 Exchange發送消息時，也必須指定消息的 RoutingKey。
Exchange不再把消息交給每一個綁定的隊列，而是根據消息的Routing Key進行判斷，只有隊列的Routingkey與消息的 Routing key完全一致，才會接收到消息

案例需求如下：

利用@RabbitListener聲明Exchange、Queue、RoutingKey
在consumer服務中，編寫兩個消費者方法，分別監聽direct.queue1和direct.queue2
在publisher中編寫測試方法，向itcast. direct發送消息

3.5.1.基於註解聲明隊列和交換機

基於@Bean的方式聲明隊列和交換機比較麻煩，Spring還提供了基於註解方式來聲明。

在consumer的SpringRabbitListener中添加兩個消費者，同時基於註解來聲明隊列和交換機：

Quarkus 文檔

Spring Framework 官方文檔

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue1"),
    exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消費者接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue2"),
    exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消費者接收到direct.queue2的消息：【" + msg + "】");
}

3.5.2.消息發送

在publisher服務的SpringAmqpTest類中添加測試方法：

@Test
public void testSendDirectExchange() {
    // 交換機名稱
    String exchangeName = "itcast.direct";
    // 消息

spring , SSH , 後端 , JAVA

小帥説技術日志

@xiaoshuai456

标签

後端 (616)

JAVA (556)

spring (87)

SSH (37)

动态

【Spring開發】SpringCloud服務端基礎框架第4篇：3.2.WorkQueue,3.3.發佈/訂閲【附代碼文 - 动态详情

✨ 本教程項目亮點

🎯🎯🎯全教程總章節

🚀🚀🚀本篇主要內容

3.2.WorkQueue

3.2.1.消息發送

3.2.2.消息接收

3.2.3.測試

3.2.4.能者多勞

3.2.5.總結

3.3.發佈/訂閲

3.4.Fanout

3.4.1.聲明隊列和交換機

3.4.2.消息發送

3.4.3.消息接收

3.4.4.總結

3.5.Direct

3.5.1.基於註解聲明隊列和交換機

3.5.2.消息發送

Add a new 评论

小帥説技術 日志

@xiaoshuai456

标签

後端 (616)

JAVA (556)

spring (87)

SSH (37)

动态

【Spring開發】SpringCloud服務端基礎框架第4篇：3.2.WorkQueue,3.3.發佈/訂閲【附代碼文 - 动态 详情

✨ 本教程項目亮點

🎯🎯🎯全教程總章節

🚀🚀🚀本篇主要內容

3.2.WorkQueue

3.2.1.消息發送

3.2.2.消息接收

3.2.3.測試

3.2.4.能者多勞

3.2.5.總結

3.3.發佈/訂閲

3.4.Fanout

3.4.1.聲明隊列和交換機

3.4.2.消息發送

3.4.3.消息接收

3.4.4.總結

3.5.Direct

3.5.1.基於註解聲明隊列和交換機

3.5.2.消息發送

Add a new 评论

小帥説技術日志

【Spring開發】SpringCloud服務端基礎框架第4篇：3.2.WorkQueue,3.3.發佈/訂閲【附代碼文 - 动态详情