服务器之家：专注于VPS、云服务器配置技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - Spring Boot集成RabbitMQ以及队列模式操作

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

2022-11-15 13:34一心同学 Java教程

RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,下面这篇文章主要给大家介绍了关于Spring Boot集成RabbitMQ以及队列模式操作的相关资料,文中通过实例代码介绍的非常详细,需要的朋友可以参考下

前言

本篇博客将会通过我们的实际场景来演示如何在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）

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

生产者

项目名：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 进行发送消息。

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

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

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

 

四、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 进行推送消息：

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

查看消费者的消费信息：

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

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

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

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

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

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

 

五、路由模式（Direct）

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

生产者

项目名：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 进行发送消息。

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

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

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

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

 

六、主题模式（Topic）

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

生产者

项目名：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集成RabbitMQ以及队列模式操作

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

小结

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

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

原文链接：https://blog.csdn.net/Huang_ZX_259/article/details/123300657

  • rabbitmq
  • spring boot

    • 延伸 · 阅读

    精彩推荐
    最近更新
    编辑推荐
    阅读排行
    热门标签
    501 　 406 　 415 　 2000 　 1433 　 401 　 204 　 330 　 选择排序法 　 随机抽取 　 内存卡 　 下载器 　 十六进制字符 　 响应 　 静态工厂 　 构造函数 　 swt 　 Overview 　 对象池模式 　 参数效验 　 支票 　 金额大写转换 　 空心菱形 　 AspectJ 　 面向切面编程 　 方法名 　 网址信息 　 动态代理模式 　 线程安全 　 htmlparser 　
    © 2019-2022 服务器之家(www.tuohang.net) 版权所有关于我们联系我们版权申明网站地图