02、RabbitMQ之交换机

冷不防 2023-06-24 15:26 62阅读 0赞

一、 Exchange(交换机)的作用
在RabbitMQ中，生产者发送消息不会直接将消息投递到队列中，而是先将消息投递到交换机中，在由交换机转发到具体的队列，
队列再将消息以推送或者拉取方式给消费者进行消费

  1. 创建消息              路由键         pull/push
  2.  生产者------------>交换机------------>队列------------>消费者

在这里插入图片描述

二、 Exchange(交换机)的类型

1）直连交换机：Direct Exchange
直连交换机是一种带路由功能的交换机，一个队列会和一个交换机绑定，除此之外再绑定一个routing_key，当消息被发送的时候，需要指定一个binding_key，这个消息被送达交换机的时候，就会被这个交换机送到指定的队列里面去。同样的一个binding_key也是支持应用到多个队列中的。

在这里插入图片描述
在这里插入图片描述
2）主题交换机：Topic Exchange
弥补直连交换机的缺点！
直连交换机的routing_key方案非常简单，如果我们希望一条消息发送给多个队列，那么这个交换机需要绑定上非常多的routing_key，
假设每个交换机上都绑定一堆的routing_key连接到各个队列上。那么消息的管理就会异常地困难。

所以RabbitMQ提供了一种主题交换机，发送到主题交换机上的消息需要携带指定规则的routing_key，
主题交换机会根据这个规则将数据发送到对应的(多个)队列上。

主题交换机的routing_key需要有一定的规则，交换机和队列的binding_key需要采用*.#.*…的格式，每个部分用.分开，其中
*表示一个单词
#表示任意数量（零个或多个）单词。

  1. 示例：
  2.  队列Q1绑定键为 *.TT.*
  3.  队列Q2绑定键为TT.#
  5.  如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到 
  6.  如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到

3）扇形交换机：Fanout Exchange
扇形交换机是最基本的交换机类型，它所能做的事情非常简单———广播消息。
扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要“思考”，
所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

  1. 这个交换机没有路由键概念，就算你绑了路由键也是无视的。

4）首部交换机：Headers exchange

5）默认交换机

6）Dead Letter Exchange（死信交换机）

本文只讲前三种交换机

三、交换机的属性
除交换机类型外，在声明交换机时还可以附带许多其他的属性，其中最重要的几个分别是：
Name：交换机名称
Durability：是否持久化。如果持久性，则RabbitMQ重启后，交换机还存在
Auto-delete：当所有与之绑定的消息队列都完成了对此交换机的使用后，删掉它
Arguments：扩展参数

**四、案例：交换机的使用 **

  1. 打开linux，开启docker，开启容器，容器的创建请参照
    https://blog.csdn.net/qq_43469718/article/details/103671005
    在这里插入图片描述
    输入界面版RabbitMQ
    在这里插入图片描述

  2. 创建springcloud项目
    rabbitmq02 #主模块
    rabbitmq-provider #生产者
    rabbitmq-consumer #消费者

    给父模块添加依赖

    <?xml version=”1.0” encoding=”UTF-8”?>


    4.0.0

    org.springframework.boot
    spring-boot-starter-parent
    2.1.9.RELEASE


    com.zking
    rabbitmq02
    0.0.1-SNAPSHOT
    rabbitmq02
    pom
    Demo project for Spring Boot


    rabbitmq-provider
    rabbitmq-consumer



    1.8




    org.springframework.boot
    spring-boot-starter-amqp



    org.springframework.boot
    spring-boot-starter-test
    test



    org.springframework.amqp
    spring-rabbit-test
    test



    org.springframework.boot
    spring-boot-starter-web



    org.projectlombok
    lombok
    1.18.10
    provided







    org.springframework.boot
    spring-boot-maven-plugin

子模块pom

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>com.zking</groupId>
  7.         <artifactId>rabbitmq02</artifactId>
  8.         <version>0.0.1-SNAPSHOT</version>
  9.     </parent>
  10.     <artifactId>rabbitmq-provider</artifactId>
  11.     <version>0.0.1-SNAPSHOT</version>
  12.     <name>rabbitmq-provider</name>
  13.     <packaging>jar</packaging>
  14.     <description>Demo project for Spring Boot</description>
  16. </project>
  18. <?xml version="1.0" encoding="UTF-8"?>
  19. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  20.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  21.     <modelVersion>4.0.0</modelVersion>
  22.     <parent>
  23.         <groupId>com.zking</groupId>
  24.         <artifactId>rabbitmq02</artifactId>
  25.         <version>0.0.1-SNAPSHOT</version>
  26.     </parent>
  27.     <artifactId>rabbitmq-consumer</artifactId>
  28.     <version>0.0.1-SNAPSHOT</version>
  29.     <name>rabbitmq-consumer</name>
  30.     <packaging>jar</packaging>
  31.     <description>Demo project for Spring Boot</description>
  33. </project>

配置yml
rabbitmq-provider

  1. server:
  2.   port: 8081
  3.   servlet:
  4.     context-path: /rabbitmq-provider
  5. spring:
  6.   rabbitmq:
  7.     virtual-host: my_vhost
  8.     host: 192.168.208.130
  9.     port: 5672
  10.     username: admin
  11.     password: admin

rabbitmq-consumer

  1. server:
  2.   port: 8082
  3.   servlet:
  4.     context-path: /rabbitmq-consumer
  5. spring:
  6.   rabbitmq:
  7.     virtual-host: my_vhost
  8.     host: 192.168.208.130
  9.     port: 5672
  10.     username: admin
  11.     password: admin

在这里插入图片描述
配置完毕后

1.演示直连交换机：Direct Exchange
在生产者rabbitmq-provider新建包rabbitmq ，这个包专门用来放配置类
新建类RabbitDirect 配置类

  1. package com.zking.rabbitmqprovider.rabbitmq;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.DirectExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  9. /* * 功能描述: <br> * 〈〉用于绑定队列交换机路由键的关系 * 示例：直接交换机（Driect Exchange） * @Param: * @Return: * @Author: 骄傲的骨傲天 * @Date: 2019/12/23 22:27 */
  10. @Configuration
  11. public class RabbitDirect { 
  13.     /* * 功能描述: <br> * 〈〉1.定义一队列 * @Param: * @Return: * @Author: 骄傲的骨傲天 * @Date: 2019/12/23 22:34 */
  14.     @Bean
  15.     public Queue queue(){ 
  16.         return new Queue("queue-1",true);
  17.     }
  18.     /* * 功能描述: <br> * 〈〉2.定义直接交换机 * @Param: * @Return: * @Author: 骄傲的骨傲天 * @Date: 2019/12/23 22:34 */
  19.     @Bean
  20.     public DirectExchange directExchange(){ 
  21.         return new DirectExchange("direct-exchange");
  22.     }
  24.     /* * 功能描述: <br> * 〈〉3.将队列与直连交换机进行绑定，并以路由键来串联 * @Param: * @Return: * @Author: 骄傲的骨傲天 * @Date: 2019/12/23 22:36 */
  25.     @Bean
  26.     public Binding directBinding(){ 
  27.         return BindingBuilder.bind(queue()).
  28.                 to(directExchange()).with("direct-routing-key");
  29.     }
  30. }

新建controller包与rabbitmq同级目录
新建类SendController用户来向交换机发送消息根据路由键传到消息队列中
此类头上有个@RestController注解

  1. @Autowired
  2.     private RabbitTemplate rabbitTemplate;
  4.      @RequestMapping("/senderByDricet")
  5.     public Map<String,Object> senderByDricet(){ 
  6.         Map<String,Object> map=new HashMap<String,Object>();
  8.         Map<String, Object> data = this.createData();
  9.         log.info("生产者发送消息，queue={},directExchange={},routingkey={}",
  10.                 "queue-1","direct-exchange","direct-routing-key");
  11.         log.info("msg={}",data);
  13.         //第一个参数：交换机的名称
  14.         //第二个参数：路由键
  15.         //第三个参数：数据
  16.         rabbitTemplate.convertAndSend("direct-exchange","direct-routing-key",data);
  18.         map.put("code",1);
  19.         map.put("msg","消息发送成功");
  21.         return map;
  22.     }
  24.     public Map<String,Object> createData(){ 
  25.         Map<String,Object> data=new HashMap<String,Object>();
  26.         String ceateDate= LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
  27.         String message="hello,rabbitmq!!";
  29.         data.put("createDate",ceateDate);
  30.         data.put("massage",message);
  31.         return data;
  32.     }

运行
postman测试一下
在这里插入图片描述
再看下RabbitMQ web版有没有此数据列表
在这里插入图片描述
可以看到名为queue-1的消息队列中是有一条消息的

往消费者rabbitmq-consumer写入代码，消费此消息
创建rabbitmq包
创建RabbitDirectReceiver类
开始消费

  1. package com.zking.rabbitmqconsumer.rabbitmq;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  6. import org.springframework.stereotype.Component;
  8. import java.util.Map;
  10. @Component
  11. @RabbitListener(queues = { "queue-1"})
  12. @Slf4j
  13. public class RabbitDirectReceiver { 
  15.     @RabbitHandler
  16.     public void handlerMessage(Map<String,Object> map){ 
  17.         log.info("消费者接收消息。。。");
  18.         log.info("msg={}",map);
  19.     }
  21. }

在这里插入图片描述
看下消息已经被消费了在这里插入图片描述
关闭消费者rabbitmq-consumer，不让进行消费，因为演示主题交换机需要看到效果

2.演示主题交换机：Topic Exchange

同理我们先来制造消息
在生产者rabbitmq-provider中的rabbitmq包下创建主题交换机配置类RabbitTopicConfig

  1. package com.zking.rabbitmqprovider.rabbitmq;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.Queue;
  6. import org.springframework.amqp.core.TopicExchange;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  10. /* * 功能描述: <br> * 〈〉主题交换器 * @Param: * @Return: * @Author: 骄傲的骨傲天 * @Date: 2019/12/23 23:33 */
  11. @Configuration
  12. public class RabbitTopicConfig { 
  14.     @Bean
  15.     public Queue queueX(){ 
  16.         return new Queue("queue-x",true);
  17.     }
  19.     @Bean
  20.     public Queue queueY(){ 
  21.         return new Queue("queue-y",true);
  22.     }
  24.     @Bean
  25.     public Queue queueZ(){ 
  26.         return new Queue("queue-z",true);
  27.     }
  29.     @Bean
  30.     public TopicExchange topicExchange(){ 
  31.         return new TopicExchange("topic-exchange");
  32.     }
  34.     @Bean
  35.     public Binding bindingX(){ 
  36.         return BindingBuilder.bind(queueX()).to(topicExchange())
  37.                 .with("topic.person.xxx");
  38.     }
  40.     @Bean
  41.     public Binding bindingY(){ 
  42.         return BindingBuilder.bind(queueY()).to(topicExchange())
  43.                 .with("topic.person.yyy");
  44.     }
  46.     @Bean
  47.     public Binding bindingZ(){ 
  48.         return BindingBuilder.bind(queueZ()).to(topicExchange())
  49.                 .with("topic.person.*");
  50.     }
  51. }

可以看到我创建了三个消息队列：queue-x queue-y queue-z
注意看bindingZ方法的路由键规则topic.person.*
主题交换机的routing_key需要有一定的规则，交换机和队列的binding_key需要采用*.#.*.....的格式，每个部分用.分开，其中
*表示一个单词
#表示任意数量（零个或多个）单词。

然后再controller包下的SendController类中添加senderByTopic方法(演示直连交换机所创建的)

  1. @RequestMapping("/senderByTopic")
  2.     public Map<String,Object> senderByTopic(String routingKey){ 
  3.         Map<String,Object> map=new HashMap<String,Object>();
  5.         Map<String, Object> data = this.createData();
  6.         log.info("生产者发送消息，directExchange={},routingkey={}",
  7.                 "topic-exchange",routingKey);
  8.         log.info("msg={}",data);
  10.         //第一个参数：交换机的名称
  11.         //第二个参数：路由键
  12.         //第三个参数：数据
  13.         rabbitTemplate.convertAndSend("topic-exchange",
  14.                 routingKey,data);
  16.         map.put("code",1);
  17.         map.put("msg","消息发送成功");
  19.         return map;
  20.     }

此方法中有个routingKey，这就是路由键，是活的，从前端传过的

重启rabbitmq-provider，使用postman测试一下
在这里插入图片描述
结果：
在这里插入图片描述
可以发现我的路由键参数为：topic.person.yyy 同时匹配queue-y和queue-z的规则，所以两个消息队列中都有消息

往消费者rabbitmq-consumer写入代码，消费此两条消息
在rabbitmq包下创建三个类：RabbitTopicReceiverX RabbitTopicReceiverY RabbitTopicReceiverZ

  1. package com.zking.rabbitmqconsumer.rabbitmq;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  9. import java.util.Map;
  11. @Component
  12. @RabbitListener(queues = { "queue-x"})
  13. @Slf4j
  14. public class RabbitTopicReceiverX { 
  15.     @RabbitHandler
  16.     public void handlerMessage(Map<String,Object> map){ 
  17.         log.info("消费者接收消息。。。");
  18.         log.info("RabbitTopicReceiverX.handlermessage={}",map);
  19.     }
  20. }
  22. package com.zking.rabbitmqconsumer.rabbitmq;
  25. import lombok.extern.slf4j.Slf4j;
  26. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  27. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  28. import org.springframework.stereotype.Component;
  30. import java.util.Map;
  32. @Component
  33. @RabbitListener(queues = { "queue-y"})
  34. @Slf4j
  35. public class RabbitTopicReceiverY { 
  36.     @RabbitHandler
  37.     public void handlerMessage(Map<String,Object> map){ 
  38.         log.info("消费者接收消息。。。");
  39.         log.info("RabbitTopicReceiverY.handlermessage={}",map);
  40.     }
  41. }
  43. package com.zking.rabbitmqconsumer.rabbitmq;
  46. import lombok.extern.slf4j.Slf4j;
  47. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  48. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  49. import org.springframework.stereotype.Component;
  51. import java.util.Map;
  53. @Component
  54. @RabbitListener(queues = { "queue-z"})
  55. @Slf4j
  56. public class RabbitTopicReceiverZ { 
  57.     @RabbitHandler
  58.     public void handlerMessage(Map<String,Object> map){ 
  59.         log.info("消费者接收消息。。。");
  60.         log.info("RabbitTopicReceiverZ.handlermessage={}",map);
  61.     }
  62. }

运行消费者
在这里插入图片描述
在这里插入图片描述
可以看到 y z消息队列已经被消费了

3.演示扇形交换机
扇形交换机是与路由键无关的
首先在生产者rabbitmq-provider下的rabbitmq中创建扇形交换机的配置类RabbitFanoutConfig

  1. package com.zking.rabbitmqprovider.rabbitmq;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.FanoutExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  10. import javax.validation.executable.ValidateOnExecution;
  12. /* * 功能描述: <br> * 〈〉扇形交换机，用于广播消息，于路由键无关 * @Param: * @Return: * @Author: 骄傲的骨傲天 * @Date: 2019/12/23 23:32 */
  13. @Configuration
  14. public class RabbitFanoutConfig { 
  16.     @Bean
  17.     public Queue queueA(){ 
  18.         return new Queue("queue-a",true);
  19.     }
  21.     @Bean
  22.     public Queue queueB(){ 
  23.         return new Queue("queue-b",true);
  24.     }
  26.     @Bean
  27.     public Queue queueC(){ 
  28.         return new Queue("queue-c",true);
  29.     }
  31.     @Bean
  32.     public FanoutExchange fanoutExchange(){ 
  33.         return new FanoutExchange("fanout-exchange");
  34.     }
  36.     @Bean
  37.     public Binding bindingA(){ 
  38.         return BindingBuilder.bind(queueA()).to(fanoutExchange());
  39.     }
  41.     @Bean
  42.     public Binding bindingB(){ 
  43.         return BindingBuilder.bind(queueB()).to(fanoutExchange());
  44.     }
  46.     @Bean
  47.     public Binding bindingC(){ 
  48.         //可以看到这里不需要with
  49.         return BindingBuilder.bind(queueC()).to(fanoutExchange());
  50.     }
  51. }

在controller包中的SendController中添加方法

  1. @RequestMapping("/senderByFanout")
  2.     public Map<String,Object> senderByFanout(){ 
  3.         Map<String,Object> map=new HashMap<String,Object>();
  5.         Map<String, Object> data = this.createData();
  6.         log.info("生产者发送消息，fanoutExchange={}",
  7.                 "fanout-exchange");
  8.         log.info("msg={}",data);
  10.         //第一个参数：交换机的名称
  11.         //第二个参数：路由键
  12.         //第三个参数：数据
  13.         rabbitTemplate.convertAndSend("fanout-exchange",
  14.                 null,data);
  16.         map.put("code",1);
  17.         map.put("msg","消息发送成功");
  19.         return map;
  20.     }

可以看到rabbitTemplate.convertAndSend("fanout-exchange", null,data);路由键是null
因此，此方法发送消息所有与fanout-exchange交换机关联的消息队列都会被发送消息

运行生产者rabbitmq-provider
postman测试一下
在这里插入图片描述
在这里插入图片描述
三个消息队列都有消息

往消费者rabbitmq-consumer写入代码，消费消息
在rabbitmq包下写入三个类：RabbitFanoutReceiverA RabbitFanoutReceiverB RabbitFanoutReceiverC

  1. package com.zking.rabbitmqconsumer.rabbitmq;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  6. import org.springframework.stereotype.Component;
  8. import java.util.Map;
  10. @Component
  11. @RabbitListener(queues = { "queue-a"})
  12. @Slf4j
  13. public class RabbitFanoutReceiverA { 
  14.     @RabbitHandler
  15.     public void handlerMessage(Map<String,Object> map){ 
  16.         log.info("消费者接收消息。。。");
  17.         log.info("RabbitTopicReceiverA.handlermessage={}",map);
  18.     }
  19. }
  21. package com.zking.rabbitmqconsumer.rabbitmq;
  23. import lombok.extern.slf4j.Slf4j;
  24. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  25. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  26. import org.springframework.stereotype.Component;
  28. import java.util.Map;
  30. @Component
  31. @RabbitListener(queues = { "queue-b"})
  32. @Slf4j
  33. public class RabbitFanoutReceiverB { 
  34.     @RabbitHandler
  35.     public void handlerMessage(Map<String,Object> map){ 
  36.         log.info("消费者接收消息。。。");
  37.         log.info("RabbitTopicReceiverB.handlermessage={}",map);
  38.     }
  39. }
  41. package com.zking.rabbitmqconsumer.rabbitmq;
  43. import lombok.extern.slf4j.Slf4j;
  44. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  45. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  46. import org.springframework.stereotype.Component;
  48. import java.util.Map;
  50. @Component
  51. @RabbitListener(queues = { "queue-c"})
  52. @Slf4j
  53. public class RabbitFanoutReceiverC { 
  54.     @RabbitHandler
  55.     public void handlerMessage(Map<String,Object> map){ 
  56.         log.info("消费者接收消息。。。");
  57.         log.info("RabbitTopicReceiverC.handlermessage={}",map);
  58.     }
  59. }

启动消费者
在这里插入图片描述
在这里插入图片描述
效果演示完毕!!!
给大家看下项目最终的结构
在这里插入图片描述

在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，62人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Rabbitmq死信交换机

    正常的Rabbitmq流程是生产者把消息先到交换机,交换机分发到队列,然后消费者从队列中取出消息 死信交换机就是给消息设置过期时间TTL,然后将正常的队列绑定死信交换机,死信

    红太狼红太狼/ 0 赞/ 122 阅读

    相关 02、RabbitMQ之交换机

    一、 Exchange(交换机)的作用 在RabbitMQ中，生产者发送消息不会直接将消息投递到队列中，而是先将消息投递到交换机中，在由交换机转发到具体的队列， 队列再

    冷不防冷不防/ 0 赞/ 62 阅读

    相关 RabbitMQ之Exchange交换机

    RabbitMQ中的Exchange的作用 消息队列，消息通过发送和Exchange之后最终到达的地方，到达Queue的消息及进入了等待消费的状态。每个消息都会被发送到一

    迷南。迷南。/ 0 赞/ 417 阅读