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

RabbitMq的使用_rabbitmq 手动创建 exchange

作者：2023面试高手 | 2024-06-18 06:59:53

赞

踩

rabbitmq 手动创建 exchange

以提供者、消费者为例、

准备 ：

        创建Springboot项目 消费、提供模块、配置yml文件、导入依赖

消费者:

        Yaml文件

  1. server:
  2.   port: 1234
  3. spring:
  4.   rabbitmq:
  5.     host: 127.0.0.1
  6.     port: 5672
  7.     username: guest
  8.     password: guest

          Pom依赖

  1. <dependencies>
  2.  
  3.         <!--  rabbitmq 的依赖 -->
  4.         <dependency>
  5.             <groupId>org.springframework.boot</groupId>
  6.             <artifactId>spring-boot-starter-amqp</artifactId>
  7.         </dependency>
  8.  
  9.         <dependency>
  10.             <groupId>org.springframework.boot</groupId>
  11.             <artifactId>spring-boot-starter-web</artifactId>
  12.         </dependency>
  13.  
  14.         <dependency>     
  15.             <groupId>org.springframework.boot</groupId>
  16.             <artifactId>spring-boot-starter-test</artifactId>
  17.             <scope>test</scope>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.springframework.amqp</groupId>
  21.             <artifactId>spring-rabbit-test</artifactId>
  22.             <scope>test</scope>
  23.         </dependency>
  24.  
  25.     </dependencies>

提供者：

        Yaml文件

  1. server:
  2.     port: 12345
  3. spring:
  4.   rabbitmq:
  5.     host: 127.0.0.1
  6.     port: 5672
  7.     username: guest
  8.     password: guest

提供者 建立一个消息的发送接口：

  1. public interface IMessageProducerService {
  2.  
  3.     public void sendMessage1(String msg);
  4.  
  5.     public void sendMessage2(Map<String,Object> map);
  6.  
  7.     public void sendMessage3(String msg);
  8.  
  9.     public void sendDelayedMessage(String msg);
  10. }

再建立一个接口的实现类

  1. @Component
  2. @Service
  3. public class MessageProducerServiceImpl implements IMessageProducerService{
  4.     @Resource
  5.     private RabbitTemplate template;
  6.  
  7.     private void sendMessage(String exchange,String routingKey,Object message){
  8.         //exchange 交换机的名字、routingkey 路由键、message 发送的消息
  9.         //提供者发送 消息到交换机，不是消息队列、那么交换机、路由键 必须存在、否则报错
  10.         //交换机和队列每次修改**持久化**和**自动删除**等属性都要**先删除**原来交换机和队列，
  11.         //否则启动就异常，报创建**队列**失败
  12.         this.template.convertAndSend(exchange,routingKey,message);
  13.     }
  14.  
  15.     @Override
  16.     public void sendMessage1(String msg) {
  17.         this.sendMessage("dx","rk",msg);
  18.     }
  19.  
  20.     @Override
  21.     public void sendMessage2(Map<String, Object> map) {
  22.         this.sendMessage(
  23.                 "myfan","",map);
  24.     }
  25.  
  26.     @Override
  27.     public void sendMessage3(String msg) {
  28.         this.sendMessage("mytopic","a.b.log.dao",msg);
  29.         this.sendMessage("mytopic","log.dao",msg);
  30.         this.sendMessage("mytopic","log.d.c",msg);
  31.         this.sendMessage("mytopic","a.log.c",msg);
  32.     }
  33. }

消费者通过代码创建 交换机、消息队列、路由键、设置消息队列过期时间

一定要先启动消费者、通过@Bean 和@Binding 创建交换机、消息队列和绑定路由

DirectExchange 直连模式、FanoutExchange 广播模式、TopicExchange、

  1. @Configuration
  2. //延时队列只能通过代码创建绑定或Web端手动创建绑定
  3. public class ConsumerConfiger {
  4.     
  5.     public static final String EXCHANGE = "dx";   // 交换空间名称
  6.     public static final String ROUTINGKEY = "rk"; // 设置路由key
  7.  
  8.  
  9.     //1、以直连的方式创建
  10.     @Primary
  11.     @Bean
  12.     public DirectExchange directExchange() {
  13.         // 要创建的交换机
  14.         // 持久化(存硬盘，重启还在，默认true)
  15.         // 自动删除(当所有绑定队列都不在使用时删除交换机)
  16.         return new DirectExchange(EXCHANGE, true, false);
  17.     }
  18.     @Bean
  19.     public Queue queue11() {
  20.         // 要创建的队列
  21.         // 持久化(保存硬盘，重启存在，默认true)
  22.         // 排他(只有创建它的连接可用，连接关闭时无论队列中有没有消息会删除队列)
  23.         // 自动删除(当队列中有消息时，无论是否排他，关闭连接都不会删除队列，此时消费者消费完消息后再断开消费者，队列会被自动删除，删除后队列消息也丢失)
  24.         // args参数设置队列最多5条消息(保留最后的5条消息，前面的全部丢失)
  25.         return new Queue("q11",true,true,false,args);
  26.     }
  27.  
  28.     @Bean
  29.     public Queue queue12() {
  30.         // args参数设置队列过期时间10秒，10秒后所有消息清空
  31.         Map<String,Object> args = new HashMap<>();
  32.         args.put("x-message-ttl",10000);
  33.         return new Queue("q12", true,false,false,args);
  34.     }
  35.  
  36.     //1、看binding 就知道是绑定的意思、绑定 exchange 交换机 下面的 queue 消息队列 下面的        
  37.     // routingkey
  38.     @Bean
  39.     public Binding bindingExchangeQueue11(DirectExchange exchange, Queue queue11) {
  40.         return BindingBuilder.bind(queue1).to(exchange).with(ROUTINGKEY);
  41.     }
  42.     @Bean
  43.     public Binding bindingExchangeQueue12(DirectExchange exchange, Queue queue12) {
  44.         return BindingBuilder.bind(queue2).to(exchange).with(ROUTINGKEY);
  45.     }
  46.  
  47.  
  48.     //2、以广播模式创建、
  49.     /**
  50.      * FanoutExchange 广播模式
  51.      * @return
  52.      */
  53.     @Bean
  54.     public FanoutExchange getFanoutExchange(){
  55.         return new FanoutExchange("myfan");
  56.     }
  57.     @Bean
  58.     public Queue queue21() { 
  59.         return new Queue("q21");
  60.     }
  61.     @Bean
  62.     public Queue queue22() { 
  63.         return new Queue("q22");
  64.     }
  65.     @Bean
  66.     public Binding bindingExchangeQueue21(FanoutExchange exchange, Queue queue21) {
  67.         return BindingBuilder.bind(queue21).to(exchange);
  68.     }
  69.     @Bean
  70.     public Binding bindingExchangeQueue22(FanoutExchange exchange, Queue queue22) {
  71.         return BindingBuilder.bind(queue22).to(exchange);
  72.     }
  73.     
  74.  
  75.     //3、TopicExchange
  76.     // "*"用于匹配一个单词，比如"a","abc"等；
  77.     // "#"用于匹配0个或者多个单词，比如"", "abc", "abc.def"等
  78.     @Bean
  79.     public TopicExchange getTopicExchange(){
  80.         return new TopicExchange("topic");
  81.     }
  82.     @Bean
  83.     public Queue queue31() { 
  84.         return new Queue("q31");
  85.     }
  86.     @Bean
  87.     public Queue queue32() {
  88.         return new Queue("q32");
  89.     }
  90.     @Bean
  91.     public Binding bindingExchangeQueue31(TopicExchange exchange, Queue queue31) {
  92.         return BindingBuilder.bind(queue31).to(exchange).with("#.log.*");
  93.     }
  94.     @Bean
  95.     public Binding bindingExchangeQueue32(TopicExchange exchange, Queue queue32) {
  96.         return BindingBuilder.bind(queue32).to(exchange).with("*.log.#");
  97.     }
  98.  
  99. }

消费者实现监听处理类：、负责接收提供者发送的消息、

  1. @Service
  2. public class MessageConsumer {
  3.     
  4.     @RabbitListener(queues = "q11")
  5.     public void receiveMessage1(String msg){
  6.         System.out.println("【*** 接收消息11 ***】:"+msg);
  7.     }
  8.  
  9.     //注解方式绑定交换机和队列 无需创建配置类、
  10.     @RabbitListener(bindings = @QueueBinding(
  11.             value = @Queue(value = "q11"),
  12.             exchange = @Exchange(value = "dx",type = ExchangeTypes.DIRECT)
  13.     ))
  14.     //public void receiveMessage(String msg) {    // 进行消息接收处理
  15.       //System.out.println("【*** q1接收消息 ***】" + msg);
  16.     //}
  17.  
  18.  
  19.     @RabbitListener(queues = "q12")
  20.     public void receiveMessage2(String msg){
  21.         System.out.println("【*** 接收消息12 ***】:"+msg);
  22.     }   
  23.     
  24.     @RabbitListener(queues = {"q21","q22"})
  25.     public void receiveMessage212(Map<String, Object> map){
  26.         System.out.println("q21-22:"+map);
  27.     }
  28. }

最后一步、提供者创建Test测试类、或者是像我一样创建控制层、调用方法使用

  1. @RestController
  2. public class MessageController {
  3.  
  4.     @Resource
  5.     private IMessageProducerService service;
  6.  
  7.     @Resource
  8.     private RabbitTemplate template;
  9.  
  10.  
  11.  
  12.     @RequestMapping("/test1")
  13.     public void testMessage1() throws InterruptedException {
  14.         service.sendMessage1("我发的第一条消息！"); 异常问题待会说
  15.         /*
  17.             for (int i=0;i<100;i++){  //有5条多余+10条超时的进入死信队列
  18.               Thread.sleep(100);
  19.               this.service.sendMessage1("直连消息:"+i);
  20.             }
  21.         */
  22.         System.out.println("消息发送完毕");
  23.     }
  24.  
  25.     @RequestMapping("/test2")
  26.     public void testMessage2() {
  27.         Map<String,Object> map = new HashMap<>();
  28.         map.put("sid",1);
  29.         map.put("nums",2);
  30.         this.service.sendMessage2(map);
  31.         System.out.println("消息发送完毕");
  32.     }
  33.  
  34.     @RequestMapping("/test3")
  35.     public void testMessage3() {
  36.         this.service.sendMessage3("我发的topic交换机消息！");
  37.         System.out.println("消息发送完" +
  38.                 "毕");
  39.     }
  40.  
  41.  
  42.     @RequestMapping("/test4")
  43.     public void testMessage4() {
  44.         this.service.sendDelayedMessage("我发的延时交换机消息！");
  45.         System.out.println("消息发送完" +
  46.                 "毕");
  47.     }
  48. }

以上是最基本的RabbitMq的使用 、如果有异常会出现 消息队列没有接收到消息、导致方法无线循环、我们可以配置 出现异常、只出现指定次数、

消费者配置yml 文件：

  1. spring:
  2.   rabbitmq:     
  3.     #publisher-confirms: true  # 发送确认    版本似乎不再支持
  4.     #publisher-returns: true   # 路由失败回调 
  5.     template:
  6.       # 必须设置成true 消息路由失败通知监听者，false 将消息丢弃
  7.       mandatory: true
  8.     listener:
  9.       simple:
  10.         retry:
  11.           enabled: true
  12.           max-attempts: 3   # 重试3次        
  13.         prefetch: 1			# 每次从RabbitMQ获取的消息数量
  14.         default-requeue-rejected: false        
  15.         concurrency: 1		# 每个队列启动的消费者数量        
  16.         max-concurrency: 5	# 每个队列最大的消费者数量        
  17.         acknowledge-mode: manual # 签收模式为手动签收，需要在代码中手动ACK

在消费者的监听处理类 随便找个方法 加上 int i = 10 / 0; 就进入异常、最后发现最多执行三次异常、

接下来： RabbitMq延时队列的使用

消费者配置类

  1. @Configuration
  2. public class ConsumerConfig {
  3.    /**
  4. 	 *
  5.      * 延迟交换机(x-delayed-type、x-delayed-message固定写法)
  6.      */
  7. 	@Bean("delayExchange")
  8. 	public CustomExchange delayExchange(){
  9. 		Map<String,Object> args = new HashMap<>();
  10.         //x-delayed-type 固定写法 
  11. 		args.put("x-delayed-type","direct");
  12. 		return new CustomExchange("delay.exchange","x-delayed-message",true,true,args);
  13. 	}
  14.     
  15.     @Bean("delayQueue")
  16.     public Queue delayQueue(){
  17.         return new Queue("delay.queue",true,false,false);
  18.     }
  19.  
  20.     @Bean
  21.     public Binding bindingDelayExchangeQueue(CustomExchange delayExchange, Queue delayQueue){
  22.         return BindingBuilder.bind(delayQueue).to(delayExchange).with("delay.routkey").noargs();
  23.     }
  24. }

消费者监听消息队列类

  1. @Component
  2. public class MessageConsumer {
  3.    /**
  4.      * 延时队列
  5.      * @param channel
  6.      * @param message
  7.      */
  8.     @RabbitListener(queues = "delay.queue")
  9.     public void receiveDelay(Channel channel, Message message){
  10.         System.out.println("receiveDelay:" + message);
  11.         System.out.println("delay:" + message.getMessageProperties());
  12. 		System.out.println("message body:"+new String(message.getBody()));
  13.     }
  14.     
  15.     /**
  16.       * 直接接收延时消息
  17.       */
  18.     @RabbitListener(queues = {"delay.queue"})
  19.     public void receiveDelay(String msg){
  20.         System.out.println("delay.queue:"+msg);
  21.     }
  22. }

提供者：发送类

        

  1. @Component
  2. @Service
  3. public class MessageProducerServiceImpl implements IMessageProducerService{
  4.     @Resource
  5.     private RabbitTemplate template;
  6.  
  7.  
  8.     /**
  9.      * 发延时队列
  10.      * @param msg
  11.      */
  12.     @Override
  13.     public void sendDelayedMessage(String msg){
  14.         //延时队列 一定要下载最新对应的ez文件放入plugs 文件夹里,版本不同、没下载都无法生效
  15.         template.convertAndSend("delay.exchange","delay.routkey", msg, message -> {
  16.             // 设置过期时间
  17.             message.getMessageProperties().setDelay(5000);
  18.             return message;
  19.         });
  20.     }

这就是延时队列的使用、

往下走—— 死信队列

        

  1. //代码创建交换机队列并绑定
  2. @Configuration
  3. public class MessageConfig {
  4.  
  5.     /**
  6.      * 直连
  7.      * @return
  8.      */
  9.     @Bean
  10.     public DirectExchange directExchange(){
  11.         return new DirectExchange("direct",true,true);
  12.     }
  13.  
  14.     @Bean
  15.     public Queue queue11(){
  16.         Map<String,Object> args = new HashMap<>();
  17.         args.put("x-max-length",5);	//超过5个的进入死信队列   像我发送100次请求，那么到了q11 6个请求，有一个会进入死信队列，超过了的都得进去
  18.         args.put("x-dead-letter-exchange","deadExchange");
  19.         args.put("x-dead-letter-routing-key","deadRouting");
  20.         return new Queue("q11",true,false,false,args);
  21.     }
  22.  
  23.     @Bean
  24.     public Queue queue12(){
  25.         Map<String,Object> args = new HashMap<>();
  26.         args.put("x-message-ttl",10000); //超过10秒未消费的进入死信队列
  27.         args.put("x-dead-letter-exchange","deadExchange");
  28.         args.put("x-dead-letter-routing-key","deadRouting");
  29.         return new Queue("q12",true,false,false,args);
  30.     }
  31.  
  32. 	//死信交换机
  33.     @Bean
  34.     public DirectExchange deadExchange(){
  35.         return new DirectExchange("dead.exchange",true,false);
  36.     }
  37.  
  38.     //死信队列
  39.     @Bean
  40.     public Queue deadQueue(){
  41.         Map<String,Object> args = new HashMap<>();
  42.         args.put("x-dead-letter-exchange","deadExchange");
  43.         args.put("x-dead-letter-routing-key","deadRouting");
  44.         return new Queue("dead.Queue",true,false,false,args);
  45.     }
  46.  
  47.     @Bean
  48.     public Binding bindingExchangeQueue13() {
  49.         return BindingBuilder.bind(deadQueue())
  50.             .to(deadExchange()).with("deadRouting");
  51.     }
  52.  
  53.     @Bean
  54.     public Binding bindingExchangeQueue11(DirectExchange directExchange, Queue queue11) {
  55.         return BindingBuilder.bind(queue11)
  56.             .to(directExchange).with("rk");
  57.     }
  58.     
  59.     @Bean
  60.     public Binding bindingExchangeQueue12(DirectExchange directExchange, Queue queue12) {
  61.         return BindingBuilder.bind(queue12)
  62.             .to(directExchange).with("rk");
  63.     }
  64.  
  65. }

提供者发送消息:

        

  1. @RestController
  2. public class MessageController {
  3.  
  4.     @Resource
  5.     private IMessageProducerService service;
  6.  
  7.     @Resource
  8.     private RabbitTemplate template;
  9.  
  10.  
  11.     @RequestMapping("/test1")
  12.     public void testMessage1() throws InterruptedException {
  13. //        service.sendMessage1("我发的第一条消息！");
  14.         for (int i=0;i<100;i++){  //有5条多余+10条超时的进入死信队列
  15.             Thread.sleep(100);
  16.             this.service.sendMessage1("直连消息:"+i);
  17.         }
  18.         System.out.println("消息发送完毕");
  19.     }
  20.  
  21.     @RequestMapping("/test2")
  22.     public void testMessage2() {
  23.         Map<String,Object> map = new HashMap<>();
  24.         map.put("sid",1);
  25.         map.put("nums",2);
  26.         this.service.sendMessage2(map);
  27.         System.out.println("消息发送完毕");
  28.     }
  29.  
  30.     @RequestMapping("/test3")
  31.     public void testMessage3() {
  32.         this.service.sendMessage3("我发的topic交换机消息！");
  33.         System.out.println("消息发送完" +
  34.                 "毕");
  35.     }
  36.  
  37.  
  38.     @RequestMapping("/test4")
  39.     public void testMessage4() {
  40.         this.service.sendDelayedMessage("我发的延时交换机消息！");
  41.         System.out.println("消息发送完" +
  42.                 "毕");
  43.     }
  44. }

这就是RabbitMq的一个简单的使用，如果发现我有问题，请一定要指出来，我需要进步，我愿意改正

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

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

  

闽ICP备14008679号