经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Spring Boot » 查看文章
SpringBoot之RabbitMQ的使用方法
来源：jb51　　时间：2018/12/24 10:44:15　　对本文有异议

一 、RabbitMQ的介绍

RabbitMQ是消息中间件的一种，消息中间件即分布式系统中完成消息的发送和接收的基础软件，消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息，而消费者从消息队列中消费信息.具体过程如下：

从上图可看出，对于消息队列来说，生产者、消息队列、消费者是最重要的三个概念，生产者发消息到消息队列中去，消费者监听指定的消息队列，并且当消息队列收到消息之后，接收消息队列传来的消息，并且给予相应的处理。消息队列常用于分布式系统之间互相信息的传递。

对于RabbitMQ来说，除了这三个基本模块以外，还添加了一个模块,即交换机(Exchange)。它使得生产者和消息队列之间产生了隔离，生产者将消息发送给交换机，而交换机则根据调度策略把相应的消息转发给对应的消息队列。

交换机的主要作用是接收相应的消息并且绑定到指定的队列。交换机有四种类型，分别为Direct、topic、headers、Fanout。

Direct是RabbitMQ默认的交换机模式，也是最简单的模式。即创建消息队列的时候，指定一个BindingKey。当发送者发送消息的时候，指定对应的Key。当Key和消息队列的BindingKey一致的时候，消息将会被发送到该消息队列中。

topic转发信息主要是依据通配符，队列和交换机的绑定主要是依据一种模式(通配符+字符串)，而当发送消息的时候，只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中。

headers也是根据一个规则进行匹配，在消息队列和交换机绑定的时候会指定一组键值对规则，而发送消息的时候也会指定一组键值对规则，当两组键值对规则相匹配的时候，消息会被发送到匹配的消息队列中。

Fanout是路由广播的形式，将会把消息发给绑定它的全部队列，即便设置了key，也会被忽略。

二 、SpringBoot整合RabbitMQ(Direct模式)

SpringBoot整合RabbitMQ非常简单，首先还是pom.xml引入依赖。

  1.  <dependency>
  2.       <groupId>org.springframework.boot</groupId>
  3.       <artifactId>spring-boot-starter-amqp</artifactId>
  4.   </dependency>

在application.properties中配置RabbitMQ相关的信息，并首先启动了RabbitMQ实例，并创建两个queue。

  1. spring.application.name=spirng-boot-rabbitmq
  2. spring.rabbitmq.host=127.0.0.1
  3. spring.rabbitmq.port=5672
  4. spring.rabbitmq.username=admin
  5. spring.rabbitmq.password=admin

配置Queue(消息队列)，由于采用的是Direct模式，需要在配置Queue的时候指定一个键，使其和交换机绑定。

  1. @Configuration
  2. public class RabbitConfig {
  3.   
  4.   @Bean
  5.   public org.springframework.amqp.core.Queue Queue() {
  6.  
  7.     return new org.springframework.amqp.core.Queue("hello");
  8.  
  9.   }
  10. }

接着就可以发送消息啦。在SpringBoot中，我们使用AmqpTemplate去发送消息。代码如下：

  1. @Component
  2. public class HelloSender {
  3.   
  4.   @Autowired
  5.   private AmqpTemplate rabbitTemplate;
  6.  
  7.   public void send(int index) {
  8.  
  9.     String context = "hello Queue "+index + new Date();
  10.     System.out.println("Sender : " + context);
  11.     this.rabbitTemplate.convertAndSend("hello", context);
  12.   }
  13. }

生产者发送消息之后就需要消费者接收消息。这里定义了两个消息消费者，用来模拟生产者与消费者一对多的关系。

  1. @Component
  2. @RabbitListener(queues = "hello")
  3. public class HelloReceiver {
  4.   
  5.   @RabbitHandler
  6.   public void process(String hello) {
  7.     System.out.println("Receiver1 : " + hello);
  8.   }
  9. }
  10. @Component
  11. @RabbitListener(queues = "hello")
  12. public class HelloReceiver2 {
  13.   
  14.   @RabbitHandler
  15.   public void process(String hello) {
  16.     System.out.println("Receiver2 : " + hello);
  17.   }
  18. }

在单元测试中模拟发送消息，批量发送10条消息，两个接收者分别接收了5条消息。

  1. @Autowired
  2.   private HelloSender helloSender;
  3.   @Test
  4.   public void hello() throws Exception {
  5.     for(int i=0;i<10;i++)
  6.     {
  7.       helloSender.send(i);
  8.     }
  9.   }

实际上RabbitMQ还可以支持发送对象，当然由于涉及到序列化和反序列化，该对象要实现Serilizable接口。这里定义了User对象，用来做发送消息内容。

  1. import java.io.Serializable;
  2. public class User implements Serializable{
  3.   
  4.   private String name;
  5.   private String pwd;
  6.   
  7.   public String getPwd() {
  8.     return pwd;
  9.   }
  10.   public void setPwd(String pwd) {
  11.     this.pwd = pwd;
  12.   }
  13.   public String getName() {
  14.     return name;
  15.   }
  16.   public void setName(String name) {
  17.     this.name = name;
  18.   }
  19.   public User(String name, String pwd) {
  20.     this.name = name;
  21.     this.pwd = pwd;
  22.   }  
  23.   @Override
  24. public String toString() {
  25.     return "User{" +"name='" + name + '\'' +", pwd='" + pwd + '\'' +'}';
  26.   }
  27. }

在生产者中发送User对象。

  1. @Component
  2. public class ModelSender {
  3.   
  4.   @Autowired
  5.   private AmqpTemplate rabbitTemplate;
  6.   
  7.   public void sendModel(User user) {
  8.  
  9.     System.out.println("Sender object: " + user.toString());
  10.     this.rabbitTemplate.convertAndSend("object", user);
  11.  
  12.   }
  13. }

在消费者中接收User对象。

  1. @Component
  2. @RabbitListener(queues = "object")
  3. public class ModelRecevicer {
  4.   
  5.   @RabbitHandler
  6.   public void process(User user) {
  7.  
  8.     System.out.println("Receiver object : " + user);
  9.  
  10.   }
  11. }

在单元测试中注入ModelSender 对象，实例化User对象，然后发送。

  1. @Autowired
  2.   private ModelSender modelSender;
  3.   @Test
  4.   public void model() throws Exception {
  5.     
  6.     User user=new User("abc","123");    
  7.     modelSender.sendModel(user);
  8.   }

三 、SpringBoot整合RabbitMQ(Topic转发模式)

首先需要在RabbitMQ服务端创建交换机topicExchange，并绑定两个queue：topic.message、topic.messages。

新建TopicRabbitConfig,设置对应的queue与binding。

  1. @Configuration
  2. public class TopicRabbitConfig {
  3.   
  4.   final static String message = "topic.message";
  5.   final static String messages = "topic.messages";
  6.   
  7.   @Bean
  8.   public Queue queueMessage() {
  9.     return new Queue(TopicRabbitConfig.message);
  10.   } 
  11.   
  12.   @Bean
  13.   public Queue queueMessages() {
  14.     return new Queue(TopicRabbitConfig.messages);
  15.   }
  16.  
  17.   @Bean
  18.   TopicExchange exchange() {
  19.     return new TopicExchange("topicExchange");
  20.   }
  21.  
  22.   @Bean
  23.   Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
  24.     return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
  25.   }
  26.  
  27.   @Bean
  28.   Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
  29.     return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
  30.   }
  31. }

创建消息生产者，在TopicSender中发送3个消息。

  1. @Component
  2. public class TopicSender {
  3.   
  4.   @Autowired
  5.   private AmqpTemplate rabbitTemplate;
  6.  
  7.   public void send() {
  8.     String context = "hi, i am message all";
  9.     System.out.println("Sender : " + context);
  10.     this.rabbitTemplate.convertAndSend("topicExchange", "topic.1", context);
  11.   }
  12.  
  13.   public void send1() {
  14.     String context = "hi, i am message 1";
  15.     System.out.println("Sender : " + context);
  16.     this.rabbitTemplate.convertAndSend("topicExchange", "topic.message", context);
  17.   }
  18.  
  19.   public void send2() {    
  20.     String context = "hi, i am messages 2";
  21.     System.out.println("Sender : " + context);
  22.     this.rabbitTemplate.convertAndSend("topicExchange", "topic.messages", context);    
  23.   }
  24. }

生产者发送消息，这里创建了两个接收消息的消费者。

  1. @Component
  2. @RabbitListener(queues = "topic.message")
  3. public class TopicReceiver {
  4.   
  5.   @RabbitHandler
  6.   public void process(String message) {
  7.     System.out.println("Topic Receiver1 : " + message);
  8.   }
  9. }
  10.  
  11. @Component
  12. @RabbitListener(queues = "topic.messages")
  13. public class TopicReceiver2 {
  14.   
  15.   @RabbitHandler
  16.   public void process(String message) {
  17.  
  18.     System.out.println("Topic Receiver2 : " + message);
  19.  
  20.   }
  21. }

在单元测试中注入TopicSender，利用topicSender 发送消息。

  1. @Autowired
  2.   private TopicSender topicSender;
  3.   @Test
  4.   public void topicSender() throws Exception {      
  5.     topicSender.send();    
  6.     topicSender.send1();    
  7.     topicSender.send2();
  8.   }

从上面的输出结果可以看到，Topic Receiver2 匹配到了所有消息，Topic Receiver1只匹配到了1个消息。

四 、SpringBoot整合RabbitMQ(Fanout Exchange形式)

Fanout Exchange形式又叫广播形式，因此我们发送到路由器的消息会使得绑定到该路由器的每一个Queue接收到消息。首先需要在RabbitMQ服务端创建交换机fanoutExchange，并绑定三个queue：fanout.A、fanout.B、fanout.C。

与Topic类似，新建FanoutRabbitConfig，绑定交换机和队列。

  1. @Configuration
  2. public class FanoutRabbitConfig {
  3.   
  4.   @Bean
  5.   public Queue AMessage() {
  6.     return new Queue("fanout.A");
  7.   }
  8.   
  9.   @Bean
  10.   public Queue BMessage() {
  11.     return new Queue("fanout.B");
  12.   }
  13.  
  14.   @Bean
  15.   public Queue CMessage() {
  16.     return new Queue("fanout.C");
  17.   }
  18.  
  19.   @Bean
  20.   FanoutExchange fanoutExchange() {
  21.     return new FanoutExchange("fanoutExchange");
  22.   }
  23.  
  24.   @Bean
  25.   Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
  26.     return BindingBuilder.bind(AMessage).to(fanoutExchange);
  27.   }
  28.  
  29.   @Bean
  30.   Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
  31.     return BindingBuilder.bind(BMessage).to(fanoutExchange);
  32.   }
  33.  
  34.   @Bean
  35.   Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
  36.     return BindingBuilder.bind(CMessage).to(fanoutExchange);
  37.   }
  38. }

创建消息生产者，在FanoutSender中发送消息。

  1. @Component
  2. public class FanoutSender {
  3.  
  4.   @Autowired
  5.   private AmqpTemplate rabbitTemplate;
  6.  
  7.   public void send() {
  8.     
  9.     String context = "hi, fanout msg ";
  10.     System.out.println("FanoutSender : " + context);
  11.     this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
  12.  
  13.   }
  14. }

然后创建了3个接收者FanoutReceiverA、FanoutReceiverB、FanoutReceiverC。

  1. @Component
  2. @RabbitListener(queues = "fanout.A")
  3. public class FanoutReceiverA {
  4.   
  5.   @RabbitHandler
  6.   public void process(String message) {
  7.     System.out.println("fanout Receiver A : " + message);
  8.   }
  9. }
  10. @Component
  11. @RabbitListener(queues = "fanout.B")
  12. public class FanoutReceiverB {
  13.   
  14.   @RabbitHandler
  15.   public void process(String message) {
  16.     System.out.println("fanout Receiver B: " + message);
  17.   }
  18. }
  19. @Component
  20. @RabbitListener(queues = "fanout.C")
  21. public class FanoutReceiverC {
  22.   
  23.   @RabbitHandler
  24.   public void process(String message) {
  25.     System.out.println("fanout Receiver C: " + message);
  26.   }
  27. }

在单元测试中注入消息发送者，发送消息。

  1.  @Autowired
  2.   private FanoutSender fanoutSender;
  3.   @Test
  4.   public void fanoutSender() throws Exception {
  5.     fanoutSender.send();
  6.   }

从下图可以看到3个队列都接收到了消息。

本章节创建的类比较多，下图为本章节的结构，也可以直接查看demo源码了解。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持w3xue。

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Spring Boot热门文章

  • 史上最详 Thymeleaf 使用教程
  • Spring Boot如何通过java -jar启动
  • Spring Boot 2.x（六）：优雅的统一返回值
  • Springboot使用Logback实现日志配置与异常记录
  • 使用SpringBoot+EasyExcel+Vue实现excel表格的导入和导出详解
  • spring boot上传文件超出大小异常无法捕获问题
  • paging组件的用法和意义
  • springboot2.0集成rabbitmq的示例代码
  • Spring MVC学习step1——框架熟悉
  • spring boot ${}占位符不起作用的解决方案

    • Spring Boot推荐文章

  • 3分钟带你搞定Spring Boot中Schedule
  • 不单独部署注册中心，又要具备注册中心的功能，我能上天！
  • 从零开始学Spring Boot系列-集成Spring Security实现用户认证与授权
  • 不单独部署注册中心，又要具备注册中心的功能，咋不让我上天？
  • EasyExcel 无法读取图片？用poi写了一个工具类
  • 【进阶篇】使用 Stream 流对比两个集合的常用操作分享
  • 关于DDD和COLA的一些总结和思考
  • SnakeYaml反序列化分析
  • RestTemplate进行https请求时适配信任证书
  • Kryo反序列化链分析

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号