脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringBoot集成RabbitMQ

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

 更新时间：2022年04月20日 11:30:48   作者：一心同学  
RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,下面这篇文章主要给大家介绍了关于Spring Boot集成RabbitMQ以及队列模式操作的相关资料,文中通过实例代码介绍的非常详细,需要的朋友可以参考下

前言

本篇博客将会通过我们的实际场景来演示如何在Spring Boot中集成RabbitMQ以及如何对各种队列模式进行操作。

一、场景描述

我们通过模仿用户下订单时，订单系统分别通过短信，邮件或微信进行推送消息，如下图：

二、准备工作

（1）创建两个Spring Boot项目分别对应生产者和消费者。

（2）导入依赖。

<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）定义生产者的配置文件。

application.yml:

server:
  port: 8021
spring:
  #给项目来个名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服务器
  rabbitmq:
    host: 服务器地址
    port: 5672
    username: yixin
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /

（4）定义消费者的配置文件。

application.yml:

server:
  port: 8022
spring:
  #给项目来个名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 服务器
  rabbitmq:
    host: 服务器地址
    port: 5672
    username: yixin
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /

三、发布/订阅模式（Fanout）

🌴 生产者

项目名：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 {
 
    /**
     *  创建三个队列 ：fanout.ShortMessage   fanout.Email  fanout.WeChat
     *  将三个队列都绑定在交换机 fanoutExchange 上
     *  因为是扇型交换机, 路由键无需配置,配置也不起作用
     */
 
    @Bean
    public Queue queueShortMessage() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);
 
        //一般设置一下队列的持久化就好,其余两个就是默认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,这提供了接收/发送等等方法
    @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

邮件系统：

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("邮件系统收到消息  : " +testMessage.toString());
    }
}

短信系统：

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("短信系统收到消息  : " +testMessage.toString());
    }
}

微信系统：

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("微信系统收到消息  : " +testMessage.toString());
    }
}

现在我们已经创建好了，将rabbitmq-fanout-provider和rabbitmq-fanout-consumer项目都跑起来，在浏览器输入：http://localhost:8021/sendFanoutMessage 进行发送消息。

消费者的控制台情况如下：

四、Work模式

4.1 轮询模式

解释：所谓轮询分发就是有两个消费者监听同一个队列，那么当我们发大量消息的时候，交换器会将消息平均分配给两个消费者，就算其中一个消费者的处理效率比另一个高，也同样只能分配一样的消息数量。  

🌴 生产者

（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,这提供了接收/发送等等方法
 
    @GetMapping("/sendWorkMessage")
    public String sendDirectMessage() {
 
        for(int i=0;i<10;i++) {
 
         String message="收到消息："+i;
            //将消息携带绑定键值：shortmessage 发送到交换机DirectExchange
            rabbitTemplate.convertAndSend("WorkExchange", "", message);
 
            System.out.println("发送成功:"+i);
        }
        return "消息发送成功！";
    }
}

🌴 消费者

邮件系统：

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("邮件系统收到消息  : " +testMessage);
    }
}

短信系统：

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("短信系统收到消息  : " +testMessage);
    }
}

我们把生产者和消费者项目都启动起来，访问 http://localhost:8021/sendWorkMessage 进行推送消息：

查看消费者的消费信息：

可以发现两者的工作效率不一样，但分配到的数量确实一样的，邮件收到的是偶数，短信收到的是奇数，这就是轮询分发！ 

4.2 公平分发

解读：公平分发就是根据谁执行的效率高，那么就给其多分发消息进行处理，正所谓能者多劳。

实现公平分发很简单，在基于轮询分发的基础上，我们只需要在消费者项目的配置文件中加入以下代码：

spring:
 
    listener:
      simple:
        prefetch: 1

表示将预处理模式更改为每次读取1条消息,在消费者未回执确认之前,不在进行下一条消息的投送。

故我们的消费者的配置文件整体如下：

application.yml:

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

我们重新启动生产者和消费者项目，消费情况如下：

现在可以发现，10条消息，邮件系统消费了7条，因为邮件系统比短信系统的执行效率更高！

五、路由模式（Direct）

🌴 生产者

项目名：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 {
    /**
     *  创建三个队列 ：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);
    }
 
    //绑定  将队列和交换机绑定,并设置用于匹配键
    @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,这提供了接收/发送等等方法
 
    @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 发送到交换机DirectExchange
        rabbitTemplate.convertAndSend("DirectExchange", "shortmessage", map);
        return "消息发送成功！";
    }
}

🌴 消费者

项目名：rabbitmq-direct-consumer

邮件系统：

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("邮件系统收到消息  : " +testMessage.toString());
    }
}

短信系统：

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("短信系统收到消息  : " +testMessage.toString());
    }
}

微信系统：

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("微信系统收到消息  : " +testMessage.toString());
    }
}

我们将生产者和消费者项目都跑起来，在浏览器输入：http://localhost:8021/sendDirectMessage 进行发送消息。

我们的消费者消费情况如下：

发现确实是只有我们的短信系统收到消息了，测试成功！

六、主题模式（Topic） 

🌴 生产者

项目名：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,才会分发到该队列
    @Bean
    Binding bindingExchangeShortMessage() {
        return BindingBuilder.bind(queueShortMessage()).to(exchange()).with("topic.shortmessage");
    }
 
    //将queueEmail和TopicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
    // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
    @Bean
    Binding bindingExchangeEmail() {
        return BindingBuilder.bind(queueEmail()).to(exchange()).with("topic.#");
    }
 
    //只要是消息携带的路由键是topic.wechat,才会分发到该队列
    @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,这提供了接收/发送等等方法
    @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 发送到交换机TopicExchange
        rabbitTemplate.convertAndSend("TopicExchange", "topic.shortmessage", map);
        return "消息发送成功！";
    }
}

🌴 消费者

项目名：rabbitmq-topic-consumer

邮件系统：

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("邮件系统收到消息  : " +testMessage.toString());
    }
}

短信系统：

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("短信系统收到消息  : " +testMessage.toString());
    }
}

微信系统：

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("微信系统收到消息  : " +testMessage.toString());
    }
}

现在把我们的项目生产者和消费者项目启动起来，并输入http://localhost:8021/sendTopicMessage 进行推送消息。

对我们的消费者接受消息的情况进行查看：

可以发现只有我们的邮件和短信系统收到了通知，测试成功！

小结

以上就是【一心同学】整理的如何在【Spring Boot】中集成【RabbitMQ】并且通过【场景演示】在Spring Boot中对各种【队列模式】的使用，大家可以自己动手演示一遍，记忆更加深刻。

到此这篇关于Spring Boot集成RabbitMQ以及队列模式操作的文章就介绍到这了,更多相关SpringBoot集成RabbitMQ内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • SpringBoot环境搭建及第一个程序运行(小白教程)

    SpringBoot环境搭建及第一个程序运行(小白教程)

    这篇文章主要介绍了SpringBoot环境搭建及第一个程序运行，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2020-06-06
  • IDEA 2020 本土化,真的是全中文了(真香)

    IDEA 2020 本土化,真的是全中文了(真香)

    去年，JetBrains 网站进行了本地化，提供了 8 种不同的语言版本，而现在，团队正在对基于 IntelliJ 的 IDE 进行本地化
    2020-12-12
  • Java中判断字符串是中文或者英文的工具类分享

    Java中判断字符串是中文或者英文的工具类分享

    这篇文章主要介绍了Java中判断字符串是中文或者英文的工具类分享,本文直接给出代码,相关说明请看代码的注释,需要的朋友可以参考下
    2014-10-10
  • 如何用注解的方式实现Mybatis插入数据时返回自增的主键Id

    如何用注解的方式实现Mybatis插入数据时返回自增的主键Id

    这篇文章主要介绍了如何用注解的方式实现Mybatis插入数据时返回自增的主键Id，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2022-07-07
  • springboot实现登录功能的完整步骤

    springboot实现登录功能的完整步骤

    这篇文章主要给大家介绍了关于springboot实现登录功能的完整步骤,在web应用程序中,用户登录权限验证是非常重要的一个步骤,文中通过代码以及图文介绍的非常详细,需要的朋友可以参考下
    2023-09-09
  • Mybatis 动态表名+Map参数传递+批量操作详解

    Mybatis 动态表名+Map参数传递+批量操作详解

    这篇文章主要介绍了Mybatis 动态表名+Map参数传递+批量操作详解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2020-12-12
  • AntDesign封装全局异常处理全局拦截器

    AntDesign封装全局异常处理全局拦截器

    这篇文章主要为大家介绍了AntDesign封装全局异常处理全局拦截器,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2023-10-10
  • java struts2框架简介

    java struts2框架简介

    本文主要介绍了java struts2框架的基础知识。具有一定的参考价值，下面跟着小编一起来看下吧
    2017-01-01
  • springboot集成ftp实现文件上传

    springboot集成ftp实现文件上传

    这篇文章主要为大家详细介绍了springboot集成ftp实现文件上传，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2021-05-05
  • java实现简单的五子棋游戏

    java实现简单的五子棋游戏

    这篇文章主要为大家详细介绍了java实现简单的五子棋游戏，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2022-04-04

最新评论