devbox
Cpp五条
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Springboot整合RabbitMQ_springboot rabbitmq消费配置类

作者：Cpp五条 | 2024-06-08 04:47:45

赞

踩

springboot rabbitmq消费配置类

目录

1、创建Springboot项目、导入依赖

2、配置文件中添加rabbitmq的连接信息

3、编写rabbitmq的配置类

4、编写投递消息接口

5、编写邮件消费者

6、编写短信消费者

7、测试

8、使用两个邮件消费者监听邮件队列

9、实现能者多劳和手动消息确认

1、创建Springboot项目、导入依赖

  1.  <dependencies>
  2.         <!-- springboot-web组件 -->
  3.         <dependency>
  4.             <groupId>org.springframework.boot</groupId>
  5.             <artifactId>spring-boot-starter-web</artifactId>
  6.         </dependency>
  7.         <!-- 添加springboot对amqp的支持 -->
  8.         <dependency>
  9.             <groupId>org.springframework.boot</groupId>
  10.             <artifactId>spring-boot-starter-amqp</artifactId>
  11.         </dependency>
  12.         <dependency>
  13.             <groupId>org.apache.commons</groupId>
  14.             <artifactId>commons-lang3</artifactId>
  15.         </dependency>
  16.         <!--fastjson -->
  17.         <dependency>
  18.             <groupId>com.alibaba</groupId>
  19.             <artifactId>fastjson</artifactId>
  20.             <version>1.2.49</version>
  21.         </dependency>
  22.  
  23.         <dependency>
  24.             <groupId>org.projectlombok</groupId>
  25.             <artifactId>lombok</artifactId>
  26.         </dependency>
  27.     </dependencies>

2、配置文件中添加rabbitmq的连接信息

  1. spring:
  2.   rabbitmq:
  3.     #连接地址
  4.     host: 127.0.0.1
  5.     #端口号
  6.     port: 5672
  7.     #账号
  8.     username: guest
  9.     #密码
  10.     password: guest
  11.     #虚拟主机
  12.     virtual-host: /rkVirtualHost
  13. server:
  14.   port: 8080
  15.

3、编写rabbitmq的配置类

  1. package com.rk.config;
  2. import org.springframework.amqp.core.Binding;
  3. import org.springframework.amqp.core.BindingBuilder;
  4. import org.springframework.amqp.core.FanoutExchange;
  5. import org.springframework.amqp.core.Queue;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  8.  
  9.  
  10. /**
  11.  * @author :Rk.
  12.  * @date : 2022/10/8
  13.  * 定义完成后 rabbitmq服务器会自动创建交换机和队列 以及绑定关系
  14.  */
  15. @Configuration
  16. public class RabbitMQConfig {
  17.  
  18.  
  19.     /**
  20.      * 定义交换机名称
  21.      */
  22.     private String EXCHANGE_NAME="springboot_falout_exchange";
  23.  
  24.     /**
  25.      * 短信队列
  26.      */
  27.     private String FANOUT_SMS_QUEUE = "fanout_sms_queuerk";
  28.  
  29.     /**
  30.      * 邮件队列
  31.      */
  32.     private String FANOUT_EMAIL_QUEUE = "fanout_email_queuerk";
  33.  
  34.  
  35.     /**
  36.      * 定义 扇形交换机
  37.      */
  38.     @Bean
  39.     public FanoutExchange fanoutExchange(){
  40.         //参数 交换机名称
  41.         return new FanoutExchange(EXCHANGE_NAME);
  42.     }
  43.  
  44.     /**
  45.      *参数一:队列名称
  46.      *参数二durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
  47.      *参数三exclusive:默认也是false，是否独占队列
  48.      *参数四autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
  49.      */
  50.     @Bean
  51.     public Queue smsQueue(){
  52.         return new Queue(FANOUT_SMS_QUEUE,true, false,false);
  53.     }
  54.  
  55.     /**
  56.      *参数一:队列名称
  57.      *参数二:durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
  58.      *参数三:exclusive:默认也是false，是否独占队列
  59.      *参数四:autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
  60.      */
  61.     @Bean
  62.     public Queue emailQueue(){
  63.         return new Queue(FANOUT_EMAIL_QUEUE,true,false,false);
  64.     }
  65.  
  66.     /**
  67.      * 短信队列绑定交换机
  68.      * @param smsQueue   短信队列注入到容器的id，也就是方法名 sumQueue
  69.      * @param fanoutExchange   交换机注入到容器的id，也就是方法名 fanoutExchange
  70.      * @return
  71.      */
  72.     @Bean
  73.     public Binding bindingSmsFanoutExchange(Queue smsQueue,FanoutExchange fanoutExchange){
  74.         return  BindingBuilder.bind(smsQueue).to(fanoutExchange);
  75.     }
  76.  
  77.     /**
  78.      * 邮件队列队列绑定交换机
  79.      * @param emailQueue   短信队列注入到容器的id，也就是方法名 emailQueue
  80.      * @param fanoutExchange   交换机注入到容器的id，也就是方法名 fanoutExchange
  81.      * @return
  82.      */
  83.     @Bean
  84.     public Binding bindingEmailFanoutExchange(Queue emailQueue,FanoutExchange fanoutExchange){
  85.         return  BindingBuilder.bind(emailQueue).to(fanoutExchange);
  86.     }
  87.  
  88. }

        这里使用的是扇形交换机，交换机会把消息路由到绑定到它的所有队列。

        在Spring启动时，利用Spring Bean管理工厂BeanFactory接口，实现动态创建交换机、队列、交换机和队列的绑定关系，让我们无需进行重复的编码工作。

4、编写投递消息接口

  1. package com.rk.controller;
  2. import com.rk.domain.User;
  3. import org.springframework.amqp.core.AmqpTemplate;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.web.bind.annotation.GetMapping;
  6. import org.springframework.web.bind.annotation.RestController;
  7.  
  8. /**
  9.  * @author :Rk.
  10.  * @date : 2022/10/8
  11.  */
  12.  
  13. @RestController
  14. public class FanoutProducerController {
  15.  
  16.     @Autowired
  17.     private AmqpTemplate amqpTemplate;
  18.  
  19.     /**
  20.      * 模拟注册接口
  21.      * @param
  22.      * @return
  23.      */
  24.     @GetMapping("/registered")
  25.     public String sendMsg(){
  26.  
  27.         User user=new User("rk",22,"男","陕西西安");
  28.         //1、将user存入数据库
  29.  
  30.         //2、发送邮件和短信
  31.         //投递消息到交换机   交换机将消息路由到邮件队列和短信队列
  32.  
  33.         /**
  34.            * 参数一  交换机名称
  35.            * 参数二  路由key
  36.            * 参数三  发送内容
  37.          */
  38.         amqpTemplate.convertAndSend("springboot_falout_exchange","",user);
  39.  
  40.         return "注册完成";
  41.     }
  42.  
  43. }

        这里直接发送的user对象，必须要将user对象实现Serializable接口。

5、编写邮件消费者

  1. import com.rk.domain.User;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6.  
  7. /**
  8.  * @author :Rk.
  9.  * @date : 2022/10/8
  10. /* */
  11. @Slf4j
  12. @Component
  13. @RabbitListener(queues = "fanout_email_queuerk") //监听队列
  14. public class EmailConsumer {
  15.  
  16.     //监听到队列消息后执行
  17.     @RabbitHandler
  18.     public void process(User user) {
  19.         System.err.println("邮件消费者接收到消息发送邮件:"+user);
  20.     }
  21.  
  22. }

6、编写短信消费者

  1. import com.rk.domain.User;
  2. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  5.  
  6. /**
  7.  * @author :Rk.
  8.  * @date : 2022/10/8
  9.  */
  10.  
  11. @Component
  12. @RabbitListener(queues = "fanout_sms_queuerk")
  13. public class SumConsumer {
  14.     @RabbitHandler
  15.     public void process(User user){
  16.         System.out.println("短信消费者接收到消息发送短信:"+user);
  17.     }
  18. }

7、测试

  访问注册接口，会立即响应。

8、使用两个邮件消费者监听邮件队列

邮件消费者1:

  1. import com.rk.domain.User;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6.  
  7. /**
  8.  * @author :Rk.
  9.  * @date : 2022/10/8
  10. /* */
  11. @Slf4j
  12. @Component
  13. @RabbitListener(queues = "fanout_email_queuerk") //监听队列
  14. public class EmailConsumer1 {
  15.  
  16.  
  17.     //监听到队列消息后执行
  18.     @RabbitHandler
  19.     public void process(User user) throws InterruptedException {
  20.         //休眠10ms  模拟处理时间
  21.         Thread.sleep(10);
  22.         System.err.println("邮件消费者1接收到消息发送邮件:"+user);
  23.     }
  24.  
  25. }

邮件消费者2:

  1. import com.rk.domain.User;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Component;
  6.  
  7. /**
  8.  * @author :Rk.
  9.  * @date : 2022/10/8
  10. /* */
  11. @Slf4j
  12. @Component
  13. @RabbitListener(queues = "fanout_email_queuerk")
  14. public class EmailConsumer2 {
  15.  
  16.     @RabbitHandler
  17.     public void process(User user) throws InterruptedException {
  18.         //休眠20ms  模拟处理时间
  19.         Thread.sleep(20);
  20.         System.out.println("邮件消费者2接收到消息发送邮件:"+user);
  21.     }
  22.  
  23. }

测试: 生产者往交换机发送20条消息:

        两个消费者使用轮询的方式对消息进行消费，而且不存在重复消费。 这里消费者1的处理时间是10ms，消费者2的处理时间为20ms，所以使用轮询的方式来进行消费并不合理，需要实现能者多劳。

9、实现能者多劳和手动消息确认

更改配置文件的参数来解决这一个问题：

邮件消费者1:

  1. package com.rk.service;
  2. import com.rabbitmq.client.Channel;
  3. import com.rk.domain.User;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import java.io.IOException;
  11.  
  12. /**
  13.  * @author :Rk.
  14.  * @date : 2022/10/8
  15. /* */
  16. @Slf4j
  17. @Component
  18. @RabbitListener(queues = "fanout_email_queuerk")
  19. public class EmailConsumer1 {
  20.  
  21.     @RabbitHandler
  22.     public void process(User user, Message message, Channel channel) throws IOException {
  23.         try {
  24.              //休眠20ms  模拟处理时间
  25.             Thread.sleep(10);
  26.             System.err.println("邮件消费者1接收到消息发送邮件:"+user);
  27.             //确认一条消息
  28.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  29.         }catch (Exception e){
  30.             //消费者告诉队列信息消费失败
  31.             /**
  32.              * 拒绝确认消息:
  33.              * channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ;
  34.              * deliveryTag:该消息的index
  35.              * multiple：是否批量true:将一次性拒绝所有小于deliveryTag的消息
  36.              * requeue：被拒绝的是否重新入队列
  37.              */
  38.             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
  39.         }
  40.     }
  41. }

邮件消费者2:

  1. import com.rabbitmq.client.Channel;
  2. import com.rk.domain.User;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  8.  
  9. import java.io.IOException;
  10.  
  11. /**
  12.  * @author :Rk.
  13.  * @date : 2022/10/8
  14. /* */
  15. @Slf4j
  16. @Component
  17. @RabbitListener(queues = "fanout_email_queuerk")
  18. public class EmailConsumer4 {
  19.  
  20.     @RabbitHandler
  21.     public void process(User user, Message message, Channel channel) throws IOException {
  22.         try {
  23.             //确认一条消息
  24.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  25.             //休眠20ms  模拟处理时间
  26.             Thread.sleep(20);
  27.             System.out.println("邮件消费者2接收到消息发送邮件:"+user);
  28.         }catch (Exception e){
  29.             //消费者告诉队列信息消费失败
  30.             /**
  31.              * 拒绝确认消息:
  32.              * channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ;
  33.              * deliveryTag:该消息的index
  34.              * multiple：是否批量true:将一次性拒绝所有小于deliveryTag的消息
  35.              * requeue：被拒绝的是否重新入队列
  36.              */
  37.             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
  38.         }
  39.     }
  40.  
  41. }

        实现能者多劳和手动确认。上面的演示过程都是将生产者和消费者写在同一个项目中，在实际开发过程中，一般使用2个项目，一个项目来投递消息，一个专门用来消费消息，使用不同的JVM来运行生产者和消费者。

整合直连交换机和主题交换机可以看下面这篇博客:

Springboot 整合RabbitMq ，用心看完这一篇就够了_小目标青年的博客-CSDN博客_springboot整合rabbitmq

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Cpp五条/article/detail/688468
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号