• RabbitMQ【直连、主题、扇形交换机实战】

    目录

    1. 直连交换机（Direct实战）

         provider生产者（publisher）

         consumer消费者

    2. 主题交换机（Topic实战）

          provider生产者（publisher）

          consumer消费者

    3. 扇形交换机（Fanout实战）

          provider生产者（publisher）

          consumer消费者

    前言

       想学习RabbitMQ基础的请阅读下边博文链接

       RabbitMQ【基本使用】_JoneClassMate的博客-CSDN博客

    1.  直连交换机（Direct实战）

          provider生产者（publisher）

    • DirectConfig

    1. package com.jmh.provider.config;
    2.  
    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.  
    10. /**
    11.  * @author 蒋明辉
    12.  * @data 2022/11/25 19:02
    13.  */
    14. @Configuration
    15. @SuppressWarnings("all")
    16. public class DirectConfig {
    17.  
    18.     /**
    19.      * 创建队列
    20.      */
    21.     @Bean
    22.     public Queue directQueueA(){
    23.         return new Queue("directQueueA",true);
    24.  
    25.     }
    26.     @Bean
    27.     public Queue directQueueB(){
    28.         return new Queue("directQueueB",true);
    29.  
    30.     }
    31.     @Bean
    32.     public Queue directQueueC(){
    33.         return new Queue("directQueueC",true);
    34.  
    35.     }
    36.  
    37.     /**
    38.      * 创建交换机
    39.      */
    40.     @Bean
    41.     public DirectExchange directExchange(){
    42.         return new DirectExchange("directExchange");
    43.     }
    44.  
    45.     /**
    46.      * 设置队列和交换机的绑定
    47.      */
    48.     @Bean
    49.     public Binding bindingA(){
    50.         return BindingBuilder.bind(directQueueA()).to(directExchange()).with("AA");
    51.     }
    52.     @Bean
    53.     public Binding bindingB(){
    54.         return BindingBuilder.bind(directQueueB()).to(directExchange()).with("BB");
    55.     }
    56.     @Bean
    57.     public Binding bindingC(){
    58.         return BindingBuilder.bind(directQueueC()).to(directExchange()).with("CC");
    59.     }
    60.  
    61.  
    62. }
    •  controller
    1. package com.jmh.provider.controller;
    2.  
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:08
    11.  */
    12. @RestController
    13. @SuppressWarnings("all")
    14. public class ProviderController {
    15.  
    16.     @Autowired
    17.     private RabbitTemplate rabbitTemplate;
    18.  
    19.     /**
    20.      * 直连交换机
    21.      * @param key
    22.      * @return
    23.      */
    24.     @RequestMapping("/directSend")
    25.     public String directSend(String key){
    26.         rabbitTemplate.convertAndSend("directExchange",key,"Hello World");
    27.         return "yes";
    28.     }
    29.  
    30.  
    31.  
    32. }

         consumer消费者

    • DirectReceiverA

    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "directQueueA")
    15. @Slf4j
    16. public class DirectReceiverA {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("A接收到了"+msg);
    21.     }
    22.  
    23. }
    •  DirectReceiverB
    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "directQueueB")
    15. @Slf4j
    16. public class DirectReceiverB {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("B接收到了"+msg);
    21.     }
    22. }
    •  DirectReceiverC
    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "directQueueC")
    15. @Slf4j
    16. public class DirectReceiverC {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("C接收到了"+msg);
    21.     }
    22. }

    2. 主题交换机（Topic实战）  

     

          provider生产者（publisher）

    • TopicConfig

    1. package com.jmh.provider.config;
    2.  
    3. import org.springframework.amqp.core.*;
    4. import org.springframework.context.annotation.Bean;
    5. import org.springframework.context.annotation.Configuration;
    6.  
    7. /**
    8.  * @author 蒋明辉
    9.  * @data 2022/11/25 19:02
    10.  */
    11. @Configuration
    12. @SuppressWarnings("all")
    13. public class TopicConfig {
    14.  
    15.     private static final String KEY_A="*.a.*";
    16.     private static final String KEY_B="*.*.a";
    17.     private static final String KEY_C="a.#";
    18.  
    19.     /**
    20.      * 创建队列
    21.      */
    22.     @Bean
    23.     public Queue topicQueueA(){
    24.         return new Queue("topicQueueA",true);
    25.  
    26.     }
    27.     @Bean
    28.     public Queue topicQueueB(){
    29.         return new Queue("topicQueueB",true);
    30.  
    31.     }
    32.     @Bean
    33.     public Queue topicQueueC(){
    34.         return new Queue("topicQueueC",true);
    35.  
    36.     }
    37.  
    38.     /**
    39.      * 创建交换机
    40.      */
    41.     @Bean
    42.     public TopicExchange topicExchange(){
    43.         return new TopicExchange("topicExchange");
    44.     }
    45.  
    46.     /**
    47.      * 设置队列和交换机的绑定
    48.      */
    49.     @Bean
    50.     public Binding topicBindingA(){
    51.         return BindingBuilder.bind(topicQueueA()).to(topicExchange()).with(KEY_A);
    52.     }
    53.     @Bean
    54.     public Binding topicBindingB(){
    55.         return BindingBuilder.bind(topicQueueB()).to(topicExchange()).with(KEY_B);
    56.     }
    57.     @Bean
    58.     public Binding topicBindingC(){
    59.         return BindingBuilder.bind(topicQueueC()).to(topicExchange()).with(KEY_C);
    60.     }
    61.  
    62.  
    63. }
    •  controller
    1. package com.jmh.provider.controller;
    2.  
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:08
    11.  */
    12. @RestController
    13. @SuppressWarnings("all")
    14. public class ProviderController {
    15.  
    16.     @Autowired
    17.     private RabbitTemplate rabbitTemplate;
    18.  
    19.  
    20.     /**
    21.      * 主题交换机
    22.      * @param key
    23.      * @return
    24.      */
    25.     @RequestMapping("/topicSend")
    26.     public String topicSend(String key){
    27.         rabbitTemplate.convertAndSend("topicExchange",key,"Hello World");
    28.         return "yes";
    29.     }
    30.  
    31.  
    32.  
    33. }

          consumer消费者

    • TopicReceiverA

    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "topicQueueA")
    15. @Slf4j
    16. public class TopicReceiverA {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("A接收到了"+msg);
    21.     }
    22.  
    23. }
    •  TopicReceiverB
    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "topicQueueB")
    15. @Slf4j
    16. public class TopicReceiverB {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("B接收到了"+msg);
    21.     }
    22.  
    23. }
    •  TopicReceiverC
    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "topicQueueC")
    15. @Slf4j
    16. public class TopicReceiverC {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("C接收到了"+msg);
    21.     }
    22.  
    23. }

    3. 扇形交换机（Fanout实战）

          provider生产者（publisher）

    • FanoutConfig

    1. package com.jmh.provider.config;
    2.  
    3. import org.springframework.amqp.core.*;
    4. import org.springframework.context.annotation.Bean;
    5. import org.springframework.context.annotation.Configuration;
    6.  
    7. /**
    8.  * @author 蒋明辉
    9.  * @data 2022/11/25 19:02
    10.  */
    11. @Configuration
    12. @SuppressWarnings("all")
    13. public class FanoutConfig {
    14.  
    15.     /**
    16.      * 创建队列
    17.      */
    18.     @Bean
    19.     public Queue fanoutQueueA(){
    20.         return new Queue("fanoutQueueA",true);
    21.  
    22.     }
    23.     @Bean
    24.     public Queue fanoutQueueB(){
    25.         return new Queue("fanoutQueueB",true);
    26.  
    27.     }
    28.     @Bean
    29.     public Queue fanoutQueueC(){
    30.         return new Queue("fanoutQueueC",true);
    31.  
    32.     }
    33.  
    34.     /**
    35.      * 创建交换机
    36.      */
    37.     @Bean
    38.     public FanoutExchange fanoutExchange(){
    39.         return new FanoutExchange("fanoutExchange");
    40.     }
    41.  
    42.     /**
    43.      * 设置队列和交换机的绑定
    44.      */
    45.     @Bean
    46.     public Binding fanoutBindingA(){
    47.         return BindingBuilder.bind(fanoutQueueA()).to(fanoutExchange());
    48.     }
    49.     @Bean
    50.     public Binding fanoutBindingB(){
    51.         return BindingBuilder.bind(fanoutQueueB()).to(fanoutExchange());
    52.     }
    53.     @Bean
    54.     public Binding fanoutBindingC(){
    55.         return BindingBuilder.bind(fanoutQueueC()).to(fanoutExchange());
    56.     }
    57.  
    58.  
    59. }
    •  controller
    1. package com.jmh.provider.controller;
    2.  
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:08
    11.  */
    12. @RestController
    13. @SuppressWarnings("all")
    14. public class ProviderController {
    15.  
    16.     @Autowired
    17.     private RabbitTemplate rabbitTemplate;
    18.  
    19.  
    20.     /**
    21.      * 扇形交换机
    22.      * @param key
    23.      * @return
    24.      */
    25.     @RequestMapping("/fanoutSend")
    26.     public String fanoutSend(){
    27.         rabbitTemplate.convertAndSend("fanoutExchange",null,"Hello World");
    28.         return "yes";
    29.     }
    30.  
    31.  
    32. }

          consumer消费者

    • FanoutReceiverA

    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "fanoutQueueA")
    15. @Slf4j
    16. public class FanoutReceiverA {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("A接收到了"+msg);
    21.     }
    22.  
    23. }
    • FanoutReceiverB 
    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "fanoutQueueB")
    15. @Slf4j
    16. public class FanoutReceiverB {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("B接收到了"+msg);
    21.     }
    22.  
    23. }
    •  FanoutReceiverC
    1. package com.jmh.consumer.listener;
    2.  
    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;
    7.  
    8. /**
    9.  * @author 蒋明辉
    10.  * @data 2022/11/25 19:12
    11.  */
    12. @Component
    13. @SuppressWarnings("all")
    14. @RabbitListener(queues = "fanoutQueueC")
    15. @Slf4j
    16. public class FanoutReceiverC {
    17.  
    18.     @RabbitHandler
    19.     public void info(String msg){
    20.         log.info("C接收到了"+msg);
    21.     }
    22.  
    23. }

     

  • 相关阅读:
    SEO搜索引擎优化-SEO搜索引擎优化软件
    torch.jit.trace与torch.jit.script的区别
    java毕业生设计医院门诊分诊系统计算机源码+系统+mysql+调试部署+lw
    Dubbo3注册为应用级时报错“No provider available for the service XXX”
    Django 基于ORM的CURD、外键关联，请求的生命周期
    AI人工智能进阶-BERT/Transformer/LSTM/RNN原理与代码
    论文精读《OFT: Orthographic Feature Transform for Monocular 3D Object Detection》
    学习笔记——指针！
    【JavaScript流程控制-分支】
    十、Mysql - 全备份 - 根据二进制日志还原数据
  • 原文地址：https://blog.csdn.net/m0_63300795/article/details/128043807