Spring?Boot集成RabbitMQ以及隊列模式操作

更新時間：2022年04月20日 11:30:48 作者：一心同學(xué)

RabbitMQ是實現(xiàn)AMQP(高級消息隊列協(xié)議)的消息中間件的一種,下面這篇文章主要給大家介紹了關(guān)于Spring?Boot集成RabbitMQ以及隊列模式操作的相關(guān)資料,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下

前言

本篇博客將會通過我們的實際場景來演示如何在Spring Boot中集成RabbitMQ以及如何對各種隊列模式進行操作。

一、場景描述

我們通過模仿用戶下訂單時，訂單系統(tǒng)分別通過短信，郵件或微信進行推送消息，如下圖：

二、準備工作

（1）創(chuàng)建兩個Spring Boot項目分別對應(yīng)生產(chǎn)者和消費者。

（2）導(dǎo)入依賴。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

（3）定義生產(chǎn)者的配置文件。

application.yml:

server:
  port: 8021
spring:
  #給項目來個名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服務(wù)器
  rabbitmq:
    host: 服務(wù)器地址
    port: 5672
    username: yixin
    password: 123456
    #虛擬host 可以不設(shè)置,使用server默認host
    virtual-host: /

（4）定義消費者的配置文件。

application.yml:

server:
  port: 8022
spring:
  #給項目來個名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 服務(wù)器
  rabbitmq:
    host: 服務(wù)器地址
    port: 5672
    username: yixin
    password: 123456
    #虛擬host 可以不設(shè)置,使用server默認host
    virtual-host: /

三、發(fā)布/訂閱模式（Fanout）

?? 生產(chǎn)者

項目名：rabbitmq-fanout-provider

（1）編寫配置類。

package com.yixin.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class FanoutRabbitConfig {
 
    /**
     *  創(chuàng)建三個隊列 ：fanout.ShortMessage   fanout.Email  fanout.WeChat
     *  將三個隊列都綁定在交換機 fanoutExchange 上
     *  因為是扇型交換機, 路由鍵無需配置,配置也不起作用
     */
 
    @Bean
    public Queue queueShortMessage() {
        // durable:是否持久化,默認是false,持久化隊列：會被存儲在磁盤上，當消息代理重啟時仍然存在，暫存隊列：當前連接有效
        // exclusive:默認也是false，只能被當前創(chuàng)建的連接使用，而且當連接關(guān)閉后隊列即被刪除。此參考優(yōu)先級高于durable
        // autoDelete:是否自動刪除，當沒有生產(chǎn)者或者消費者使用此隊列，該隊列會自動刪除。
        //   return new Queue("TestDirectQueue",true,true,false);
 
        //一般設(shè)置一下隊列的持久化就好,其余兩個就是默認false
 
        return new Queue("fanout.ShortMessage",true);
    }
 
    @Bean
    public Queue queueEmail() {
        return new Queue("fanout.Email",true);
    }
 
    @Bean
    public Queue queueWeChat() {
        return new Queue("fanout.WeChat",true);
    }
 
    //Fanout交換機 起名：fanoutExchange
 
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
 
    //綁定  將隊列和交換機綁定
    @Bean
    Binding bindingExchangeShortMessage() {
        return BindingBuilder.bind(queueShortMessage()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeEmail() {
        return BindingBuilder.bind(queueEmail()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeWeChat() {
        return BindingBuilder.bind(queueWeChat()).to(fanoutExchange());
    }
}

（2）編寫一個Controller類進行推送消息。

package com.yixin.controller;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
 
@RestController
public class OrderController {
    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供了接收/發(fā)送等等方法
    @GetMapping("/sendFanoutMessage")
    public String sendFanoutMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "用戶成功下單了！";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }
}

?? 消費者

項目名：rabbitmq-fanout-consumer

郵件系統(tǒng)：

package com.yixin.consumer;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
@Component
@RabbitListener(queues = "fanout.Email")
public class Email {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("郵件系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

短信系統(tǒng)：

package com.yixin.consumer;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
@Component
@RabbitListener(queues = "fanout.ShortMessage")
public class ShortMessage {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("短信系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

微信系統(tǒng)：

package com.yixin.consumer;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
@Component
@RabbitListener(queues = "fanout.WeChat")
public class WeChat {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("微信系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

現(xiàn)在我們已經(jīng)創(chuàng)建好了，將rabbitmq-fanout-provider和rabbitmq-fanout-consumer項目都跑起來，在瀏覽器輸入：http://localhost:8021/sendFanoutMessage 進行發(fā)送消息。

消費者的控制臺情況如下：

四、Work模式

4.1 輪詢模式

解釋：所謂輪詢分發(fā)就是有兩個消費者監(jiān)聽同一個隊列，那么當我們發(fā)大量消息的時候，交換器會將消息平均分配給兩個消費者，就算其中一個消費者的處理效率比另一個高，也同樣只能分配一樣的消息數(shù)量。

?? 生產(chǎn)者

（1）編寫配置類。

package com.yixin.config;
 
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class WorkRabbitMQConfig {
    @Bean
    public Queue queueWork() {
        
        return new Queue("queue_work",true);
    }
    
    @Bean
    FanoutExchange workExchange() {
        return new FanoutExchange("WorkExchange");
    }
 
    //綁定  將隊列和交換機綁定
    @Bean
    Binding bindingExchangeShortMessage() {
        return BindingBuilder.bind(queueWork()).to(workExchange());
    }
}

（2）編寫Controller類進行推送消息。

package com.yixin.controller;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
 
@RestController
public class SendMessageController {
    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供了接收/發(fā)送等等方法
 
    @GetMapping("/sendWorkMessage")
    public String sendDirectMessage() {
 
        for(int i=0;i<10;i++) {
 
         String message="收到消息："+i;
            //將消息攜帶綁定鍵值：shortmessage 發(fā)送到交換機DirectExchange
            rabbitTemplate.convertAndSend("WorkExchange", "", message);
 
            System.out.println("發(fā)送成功:"+i);
        }
        return "消息發(fā)送成功！";
    }
}

?? 消費者

郵件系統(tǒng)：

package com.yixin.consumer;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
@Component
@RabbitListener(queues = "queue_work")
public class Email {
 
    @RabbitHandler
    public void process(String testMessage) {
 
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("郵件系統(tǒng)收到消息  : " +testMessage);
    }
}

短信系統(tǒng)：

package com.yixin.consumer;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
@Component
@RabbitListener(queues = "queue_work")
public class ShortMessage {
 
    @RabbitHandler
    public void process(String testMessage) {
 
        //休眠300毫秒，表示效率相比Email低
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("短信系統(tǒng)收到消息  : " +testMessage);
    }
}

我們把生產(chǎn)者和消費者項目都啟動起來，訪問 http://localhost:8021/sendWorkMessage 進行推送消息：

查看消費者的消費信息：

可以發(fā)現(xiàn)兩者的工作效率不一樣，但分配到的數(shù)量確實一樣的，郵件收到的是偶數(shù)，短信收到的是奇數(shù)，這就是輪詢分發(fā)！

4.2 公平分發(fā)

解讀：公平分發(fā)就是根據(jù)誰執(zhí)行的效率高，那么就給其多分發(fā)消息進行處理，正所謂能者多勞。

實現(xiàn)公平分發(fā)很簡單，在基于輪詢分發(fā)的基礎(chǔ)上，我們只需要在消費者項目的配置文件中加入以下代碼：

spring:
 
    listener:
      simple:
        prefetch: 1

表示將預(yù)處理模式更改為每次讀取1條消息,在消費者未回執(zhí)確認之前,不在進行下一條消息的投送。

故我們的消費者的配置文件整體如下：

application.yml:

server:
  port: 8022
spring:
 
  application:
    name: rabbitmq-consumer
 
  rabbitmq:
    host: 服務(wù)器地址
    port: 5672
    username: yixin
    password: 123456
 
    virtual-host: /
    listener:
      simple:
        prefetch: 1

我們重新啟動生產(chǎn)者和消費者項目，消費情況如下：

現(xiàn)在可以發(fā)現(xiàn)，10條消息，郵件系統(tǒng)消費了7條，因為郵件系統(tǒng)比短信系統(tǒng)的執(zhí)行效率更高！

五、路由模式（Direct）

?? 生產(chǎn)者

項目名：rabbitmq-direct-provider

（1）編寫配置類。

package com.yixin.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class DirectRabbitConfig {
    /**
     *  創(chuàng)建三個隊列 ：direct.ShortMessage   direct.Email  direct.WeChat
     *  將三個隊列都綁定在交換機 DirectExchange 上
     *  因為是扇型交換機, 路由鍵無需配置,配置也不起作用
     */
 
    @Bean
    public Queue queueShortMessage() {
 
        return new Queue("direct.ShortMessage",true);
    }
 
    @Bean
    public Queue queueEmail() {
        return new Queue("direct.Email",true);
    }
 
    @Bean
    public Queue queueWeChat() {
        return new Queue("direct.WeChat",true);
    }
 
    //Direct交換機 起名：DirectExchange
    @Bean
    DirectExchange DirectExchange() {
 
        return new DirectExchange("DirectExchange",true,false);
    }
 
    //綁定  將隊列和交換機綁定,并設(shè)置用于匹配鍵
    @Bean
    Binding bindingExchangeShortMessage() {
        return BindingBuilder.bind(queueShortMessage()).to(DirectExchange()).with("shortmessage");
    }
 
    @Bean
    Binding bindingExchangeEmail() {
        return BindingBuilder.bind(queueEmail()).to(DirectExchange()).with("email");
    }
 
    @Bean
    Binding bindingExchangeWeChat() {
        return BindingBuilder.bind(queueWeChat()).to(DirectExchange()).with("wechat");
    }
}

（2）編寫Controller類進行推送消息。

package com.yixin.controller;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
 
@RestController
public class SendMessageController {
    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供了接收/發(fā)送等等方法
 
    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "Direct:用戶成功下單了！";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
 
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
 
        //將消息攜帶綁定鍵值：shortmessage 發(fā)送到交換機DirectExchange
        rabbitTemplate.convertAndSend("DirectExchange", "shortmessage", map);
        return "消息發(fā)送成功！";
    }
}

?? 消費者

項目名：rabbitmq-direct-consumer

郵件系統(tǒng)：

package com.yixin.consumer;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import java.util.Map;
 
@Component
@RabbitListener(queues = "direct.Email")
public class Email {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("郵件系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

短信系統(tǒng)：

package com.yixin.consumer;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
@Component
@RabbitListener(queues = "direct.ShortMessage")
public class ShortMessage {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("短信系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

微信系統(tǒng)：

package com.yixin.consumer;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
@Component
@RabbitListener(queues = "direct.WeChat")
public class WeChat {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("微信系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

我們將生產(chǎn)者和消費者項目都跑起來，在瀏覽器輸入：http://localhost:8021/sendDirectMessage 進行發(fā)送消息。

我們的消費者消費情況如下：

發(fā)現(xiàn)確實是只有我們的短信系統(tǒng)收到消息了，測試成功！

六、主題模式（Topic）

?? 生產(chǎn)者

項目名：rabbitmq-topic-provider

（1）編寫配置類。

package com.yixin.config;
 
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class TopicRabbitConfig {
 
 
    @Bean
    public Queue queueShortMessage() {
        return new Queue("topic.ShortMessage");
    }
 
    @Bean
    public Queue queueEmail() {
        return new Queue("topic.Email");
    }
 
    @Bean
    public Queue queueWeChat() {
        return new Queue("topic.WeChat");
    }
 
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("TopicExchange");
    }
 
    //將queueShortMessage和TopicExchange綁定,而且綁定的鍵值為topic.shortmessage
    //這樣只要是消息攜帶的路由鍵是topic.shortmessage,才會分發(fā)到該隊列
    @Bean
    Binding bindingExchangeShortMessage() {
        return BindingBuilder.bind(queueShortMessage()).to(exchange()).with("topic.shortmessage");
    }
 
    //將queueEmail和TopicExchange綁定,而且綁定的鍵值為用上通配路由鍵規(guī)則topic.#
    // 這樣只要是消息攜帶的路由鍵是以topic.開頭,都會分發(fā)到該隊列
    @Bean
    Binding bindingExchangeEmail() {
        return BindingBuilder.bind(queueEmail()).to(exchange()).with("topic.#");
    }
 
    //只要是消息攜帶的路由鍵是topic.wechat,才會分發(fā)到該隊列
    @Bean
    Binding bindingExchangeWeChat() {
        return BindingBuilder.bind(queueEmail()).to(exchange()).with("topic.wechat");
    }
}

（2）編寫Controller類進行推送消息。

package com.yixin.controller;
 
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
 
@RestController
public class SendMessageController {
    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供了接收/發(fā)送等等方法
    @GetMapping("/sendTopicMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "Topic:用戶成功下單了！";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
 
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
 
        //將消息攜帶綁定鍵值：topic.shortmessage 發(fā)送到交換機TopicExchange
        rabbitTemplate.convertAndSend("TopicExchange", "topic.shortmessage", map);
        return "消息發(fā)送成功！";
    }
}

?? 消費者

項目名：rabbitmq-topic-consumer

郵件系統(tǒng)：

package com.yixin.consumer;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import java.util.Map;
 
@Component
@RabbitListener(queues = "topic.Email")
public class Email {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("郵件系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

短信系統(tǒng)：

package com.yixin.consumer;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import java.util.Map;
 
@Component
@RabbitListener(queues = "topic.ShortMessage")
public class ShortMessage {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("短信系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

微信系統(tǒng)：

package com.yixin.consumer;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import java.util.Map;
 
@Component
@RabbitListener(queues = "topic.WeChat")
public class WeChat {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("微信系統(tǒng)收到消息  : " +testMessage.toString());
    }
}

現(xiàn)在把我們的項目生產(chǎn)者和消費者項目啟動起來，并輸入http://localhost:8021/sendTopicMessage 進行推送消息。

對我們的消費者接受消息的情況進行查看：

可以發(fā)現(xiàn)只有我們的郵件和短信系統(tǒng)收到了通知，測試成功！

小結(jié)

以上就是【一心同學(xué)】整理的如何在【Spring Boot】中集成【RabbitMQ】并且通過【場景演示】在Spring Boot中對各種【隊列模式】的使用，大家可以自己動手演示一遍，記憶更加深刻。

到此這篇關(guān)于Spring Boot集成RabbitMQ以及隊列模式操作的文章就介紹到這了,更多相關(guān)SpringBoot集成RabbitMQ內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

亚洲乱码中文字幕综合,中国熟女仑乱hd,亚洲精品乱拍国产一区二区三区,一本大道卡一卡二卡三乱码全集资源,又粗又黄又硬又爽的免费视频

軟件下載

源碼下載

軟件編程

網(wǎng)絡(luò)編程

在線工具

數(shù)據(jù)庫

CMS

常用工具

Spring?Boot集成RabbitMQ以及隊列模式操作

目錄

前言

一、場景描述

二、準備工作

三、發(fā)布/訂閱模式（Fanout）

?? 生產(chǎn)者

?? 消費者

四、Work模式

4.1 輪詢模式

?? 生產(chǎn)者

?? 消費者

4.2 公平分發(fā)

五、路由模式（Direct）

?? 生產(chǎn)者

?? 消費者

六、主題模式（Topic）

?? 生產(chǎn)者

?? 消費者

小結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

Spring?Boot集成RabbitMQ以及隊列模式操作

目錄

前言

一、場景描述

二、準備工作

三、發(fā)布/訂閱模式（Fanout）

?? 生產(chǎn)者

?? 消費者

四、Work模式

4.1 輪詢模式

?? 生產(chǎn)者

?? 消費者

4.2 公平分發(fā)

五、路由模式（Direct）

?? 生產(chǎn)者

?? 消費者

六、主題模式（Topic）

?? 生產(chǎn)者

?? 消費者

小結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

一、場景描述

四、Work模式

五、路由模式（Direct）

六、主題模式（Topic）