• springboot集成rabbitmq:fanout、topic

    编写Fanout模式的消息接收

    其他模块和上文保持一致https://blog.csdn.net/weixin_59334478/article/details/127740411?spm=1001.2014.3001.5501

    ReceiveServiceImpl实现类

    1. package com.it.rabbitmq.impl;
    2.  
    3. import com.it.rabbitmq.ReceiveService;
    4. import org.springframework.amqp.core.AmqpTemplate;
    5. import org.springframework.amqp.rabbit.annotation.Exchange;
    6. import org.springframework.amqp.rabbit.annotation.Queue;
    7. import org.springframework.amqp.rabbit.annotation.QueueBinding;
    8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    9. import org.springframework.stereotype.Service;
    10.  
    11. import javax.annotation.Resource;
    12.  
    13. @Service("receiveService")
    14. public class ReceiveServiceImpl implements ReceiveService {
    15.     //注入amqp的模板类，利用这个对象来发送和接收消息
    16.     @Resource
    17.     private AmqpTemplate amqpTemplate;
    18.  
    19.  
    20.     @Override
    21.     public void receiveMessage() {
    22.         /**
    23.          * 发送消息
    24.          * 参数1为交换机名称
    25.          * 参数2位RoutingKey
    26.          * 参数3为具体发送的消息数据
    27.          */
    28.         String message= (String) amqpTemplate.receiveAndConvert("bootDirectQueue");
    29.         System.out.println(message);
    30.     }
    31.  
    32.     /**
    33.      * @RabbitListener:用于标记当前方法是一个rabbitmq的消息监听方法，作用是持续性的接收消息
    34.      * 这个方法不需要手动调用，spring会自动监听
    35.      * 属性queues:用于指定一个已经存在的队列名称，用于队列的监听
    36.      * @param message  参数就是接收到的具体消息数据
    37.      */
    38.     @RabbitListener(queues = {"bootDirectQueue"})
    39.     public void directReceive(String message) {
    40.         System.out.println("监听器接收的消息："+message);
    41.     }
    42.  
    43.     @RabbitListener(bindings = {
    44.                                  @QueueBinding(value = @Queue(),
    45.                                                 exchange = @Exchange(name="fanoutExchange",type = "fanout")
    46.                                               )})
    47.     public void fanoutReceive1(String message) {
    48.         System.out.println("fanoutReceive1监听器接收的消息："+message);
    49.     }
    50.  
    51.  
    52.     @RabbitListener(bindings = {
    53.             @QueueBinding(value = @Queue(),
    54.                     exchange = @Exchange(name="fanoutExchange",type = "fanout")
    55.             )})
    56.     public void fanoutReceive2(String message) {
    57.         System.out.println("fanoutReceive2监听器接收的消息："+message);
    58.     }
    59.  
    60. }

     

     

    编写Fanout模式的消息接收

    1.SendService接口

    1. package com.it.rabbitmq;
    2.  
    3. public interface SendService {
    4.     void sendMessage(String message);
    5.     void sendFanoutMessage(String message);
    6. }

    SendServiceImpl实现类

    1. package com.it.rabbitmq.impl;
    2.  
    3. import com.it.rabbitmq.SendService;
    4. import org.springframework.amqp.core.AmqpTemplate;
    5. import org.springframework.stereotype.Service;
    6.  
    7. import javax.annotation.Resource;
    8.  
    9. @Service("sendService")
    10. public class SendServiceImpl implements SendService {
    11.     //注入amqp的模板类，利用这个对象来发送和接收消息
    12.     @Resource
    13.     private AmqpTemplate amqpTemplate;
    14.  
    15.     @Override
    16.     public void sendMessage(String message) {
    17.         /**
    18.          * 发送消息
    19.          * 参数1为交换机名称
    20.          * 参数2位RoutingKey
    21.          * 参数3为具体发送的消息数据
    22.          */
    23.         amqpTemplate.convertAndSend("bootDirectExchange", "bootDirectRouting", message);
    24.     }
    25.  
    26.     @Override
    27.     public void sendFanoutMessage(String message) {
    28.         amqpTemplate.convertAndSend("fanoutExchange","",message);
    29.     }
    30. }

    2.RabbitMQConfig类

    1. package com.it.rabbitmq.config;
    2.  
    3. import org.springframework.amqp.core.*;
    4. import org.springframework.context.annotation.Bean;
    5. import org.springframework.context.annotation.Configuration;
    6.  
    7. @Configuration
    8. public class RabbitMQConfig {
    9.  
    10.     //配置一个direct类型的交换机
    11.     @Bean
    12.     public DirectExchange directExchange() {
    13.         return new DirectExchange("bootDirectExchange");
    14.  
    15.     }
    16.  
    17.     //配置一个队列
    18.     @Bean
    19.     public Queue directQueue() {
    20.         return new Queue("bootDirectQueue");
    21.     }
    22.  
    23.     /**
    24.      * 配置一个交换机和队列的绑定
    25.      *
    26.      * @param directQueue 需要绑定的队列对象，参数名必须要与某个@Bean的方法名完全相同这样就会自动注入
    27.      * @param directQueue 需要绑定的交换机对象，参数名必须要与某个@Bean的方法名完全相同这样就会自动注入
    28.      * @return
    29.      */
    30.     @Bean
    31.     public Binding directBinding(Queue directQueue, DirectExchange directExchange) {
    32.         //完成绑定：参数1为需要绑定的队列，参数2为需要绑定的交换机，参数3为需要绑定的routingkey
    33.         return BindingBuilder.bind(directQueue).to(directExchange).with("bootDirectRouting");
    34.     }
    35.  
    36.     //配置一个fanout类型的交换机
    37.     @Bean
    38.     public FanoutExchange fanoutExchange() {
    39.         return new FanoutExchange("fanoutExchange");
    40.     }
    41.  
    42.  
    43. }

    3.运行主函数

    1. package com.it;
    2.  
    3. import com.it.rabbitmq.SendService;
    4. import org.springframework.boot.SpringApplication;
    5. import org.springframework.boot.autoconfigure.SpringBootApplication;
    6. import org.springframework.context.ApplicationContext;
    7.  
    8. @SpringBootApplication
    9. public class RabbitmqSpringbootApplication {
    10.  
    11.     public static void main(String[] args) {
    12.  
    13.         ApplicationContext ac = SpringApplication.run(RabbitmqSpringbootApplication.class, args);
    14.         SendService sendService = (SendService) ac.getBean("sendService");
    15. //        sendService.sendMessage("boot的测试数据");
    16.  
    17.         sendService.sendFanoutMessage("boot的fanout测试数据！");
    18.  
    19.     }
    20.  
    21. }

     

     

     编写Topic模式消息接收

    ReceiveServiceImpl实现类

    1. package com.it.rabbitmq.impl;
    2.  
    3. import com.it.rabbitmq.ReceiveService;
    4. import org.springframework.amqp.core.AmqpTemplate;
    5. import org.springframework.amqp.rabbit.annotation.Exchange;
    6. import org.springframework.amqp.rabbit.annotation.Queue;
    7. import org.springframework.amqp.rabbit.annotation.QueueBinding;
    8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    9. import org.springframework.stereotype.Service;
    10.  
    11. import javax.annotation.Resource;
    12.  
    13. @Service("receiveService")
    14. public class ReceiveServiceImpl implements ReceiveService {
    15.     //注入amqp的模板类，利用这个对象来发送和接收消息
    16.     @Resource
    17.     private AmqpTemplate amqpTemplate;
    18.  
    19.  
    20.     @Override
    21.     public void receiveMessage() {
    22.         /**
    23.          * 发送消息
    24.          * 参数1为交换机名称
    25.          * 参数2位RoutingKey
    26.          * 参数3为具体发送的消息数据
    27.          */
    28.         String message= (String) amqpTemplate.receiveAndConvert("bootDirectQueue");
    29.         System.out.println(message);
    30.     }
    31.  
    32.     /**
    33.      * @RabbitListener:用于标记当前方法是一个rabbitmq的消息监听方法，作用是持续性的接收消息
    34.      * 这个方法不需要手动调用，spring会自动监听
    35.      * 属性queues:用于指定一个已经存在的队列名称，用于队列的监听
    36.      * @param message  参数就是接收到的具体消息数据
    37.      */
    38.     @RabbitListener(queues = {"bootDirectQueue"})
    39.     public void directReceive(String message) {
    40.         System.out.println("监听器接收的消息："+message);
    41.     }
    42.  
    43.     @RabbitListener(bindings = {
    44.                                  @QueueBinding(value = @Queue(),
    45.                                                 exchange = @Exchange(name="fanoutExchange",type = "fanout")
    46.                                               )})
    47.     public void fanoutReceive1(String message) {
    48.         System.out.println("fanoutReceive1监听器接收的消息："+message);
    49.     }
    50.  
    51.  
    52.     @RabbitListener(bindings = {
    53.             @QueueBinding(value = @Queue(),
    54.                     exchange = @Exchange(name="fanoutExchange",type = "fanout")
    55.             )})
    56.     public void fanoutReceive2(String message) {
    57.         System.out.println("fanoutReceive2监听器接收的消息："+message);
    58.     }
    59.  
    60.     @RabbitListener(bindings = {
    61.             @QueueBinding(value = @Queue("topic1"),
    62.                           key = {"aa"},
    63.                           exchange = @Exchange(name="topicExchange",type = "topic")
    64.             )})
    65.     public void topicReceive1(String message){
    66.         System.out.println("topic1消费者---aa---"+message);
    67.     }
    68.  
    69.     @RabbitListener(bindings = {
    70.             @QueueBinding(value = @Queue("topic2"),
    71.                     key = {"aa.*"},
    72.                     exchange = @Exchange(name="topicExchange",type = "topic")
    73.             )})
    74.     public void topicReceive2(String message){
    75.         System.out.println("topic2消费者---aa---"+message);
    76.     }
    77.  
    78.     @RabbitListener(bindings = {
    79.             @QueueBinding(value = @Queue("topic3"),
    80.                     key = {"aa.#"},
    81.                     exchange = @Exchange(name="topicExchange",type = "topic")
    82.             )})
    83.     public void topicReceive3(String message){
    84.         System.out.println("topic3消费者---aa---"+message);
    85.     }
    86.  
    87.  
    88. }

     

     

     编写Topic模式消息发送

    1.SendService接口

    1. package com.it.rabbitmq;
    2.  
    3. public interface SendService {
    4.     void sendMessage(String message);
    5.     void sendFanoutMessage(String message);
    6.     void sendTopicMessage(String message);
    7. }

    SendServiceImpl类

    1. package com.it.rabbitmq.impl;
    2.  
    3. import com.it.rabbitmq.SendService;
    4. import org.springframework.amqp.core.AmqpTemplate;
    5. import org.springframework.stereotype.Service;
    6.  
    7. import javax.annotation.Resource;
    8.  
    9. @Service("sendService")
    10. public class SendServiceImpl implements SendService {
    11.     //注入amqp的模板类，利用这个对象来发送和接收消息
    12.     @Resource
    13.     private AmqpTemplate amqpTemplate;
    14.  
    15.     @Override
    16.     public void sendMessage(String message) {
    17.         /**
    18.          * 发送消息
    19.          * 参数1为交换机名称
    20.          * 参数2位RoutingKey
    21.          * 参数3为具体发送的消息数据
    22.          */
    23.         amqpTemplate.convertAndSend("bootDirectExchange", "bootDirectRouting", message);
    24.     }
    25.  
    26.     @Override
    27.     public void sendFanoutMessage(String message) {
    28.         amqpTemplate.convertAndSend("fanoutExchange","",message);
    29.     }
    30.  
    31.     @Override
    32.     public void sendTopicMessage(String message) {
    33.         amqpTemplate.convertAndSend("topicExchange","",message);
    34.     }
    35.  
    36. }

    2.RabbitMQConfig类，提前声明一个topic的交换机

    1. package com.it.rabbitmq.config;
    2.  
    3. import org.springframework.amqp.core.*;
    4. import org.springframework.context.annotation.Bean;
    5. import org.springframework.context.annotation.Configuration;
    6.  
    7. @Configuration
    8. public class RabbitMQConfig {
    9.  
    10.     //配置一个direct类型的交换机
    11.     @Bean
    12.     public DirectExchange directExchange() {
    13.         return new DirectExchange("bootDirectExchange");
    14.  
    15.     }
    16.  
    17.     //配置一个队列
    18.     @Bean
    19.     public Queue directQueue() {
    20.         return new Queue("bootDirectQueue");
    21.     }
    22.  
    23.     /**
    24.      * 配置一个交换机和队列的绑定
    25.      *
    26.      * @param directQueue 需要绑定的队列对象，参数名必须要与某个@Bean的方法名完全相同这样就会自动注入
    27.      * @param directQueue 需要绑定的交换机对象，参数名必须要与某个@Bean的方法名完全相同这样就会自动注入
    28.      * @return
    29.      */
    30.     @Bean
    31.     public Binding directBinding(Queue directQueue, DirectExchange directExchange) {
    32.         //完成绑定：参数1为需要绑定的队列，参数2为需要绑定的交换机，参数3为需要绑定的routingkey
    33.         return BindingBuilder.bind(directQueue).to(directExchange).with("bootDirectRouting");
    34.     }
    35.  
    36.     //配置一个fanout类型的交换机
    37.     @Bean
    38.     public FanoutExchange fanoutExchange() {
    39.         return new FanoutExchange("fanoutExchange");
    40.     }
    41.  
    42.     //配置一个topic类型的交换机
    43.     @Bean
    44.     public TopicExchange topicExchange() {
    45.         return new TopicExchange("topicExchange");
    46.     }
    47.  
    48. }

    3.运行主函数

    1. package com.it;
    2.  
    3. import com.it.rabbitmq.SendService;
    4. import org.springframework.boot.SpringApplication;
    5. import org.springframework.boot.autoconfigure.SpringBootApplication;
    6. import org.springframework.context.ApplicationContext;
    7.  
    8. @SpringBootApplication
    9. public class RabbitmqSpringbootApplication {
    10.  
    11.     public static void main(String[] args) {
    12.  
    13.         ApplicationContext ac = SpringApplication.run(RabbitmqSpringbootApplication.class, args);
    14.         SendService sendService = (SendService) ac.getBean("sendService");
    15. //        sendService.sendMessage("boot的测试数据");
    16.  
    17. //        sendService.sendFanoutMessage("boot的fanout测试数据！");
    18.  
    19.         sendService.sendTopicMessage("boot的topic测试数据，key：aa");
    20.     }
    21.  
    22. }

    功能测试：

    查看接收类

    修改发送信息的routingkey

     

     修改发送信息的routingkey

     

     

  • 相关阅读:
    优盘无法格式化？分享简单解决方法！
    初始网络知识
    WPF程序打包
    倍思、南卡、漫步者开放式耳机好不好用？ 硬核测评年度最强产品
    el-menu动态加载路由，菜单的解决方案
    05 | 基础篇：某个应用的CPU使用率居然达到100%，我该怎么办？笔记
    小谈设计模式（27）—享元模式
    gitlab环境准备
    差分约束原理及其应用
    SpringCloud对服务内某个client进行单独配置
  • 原文地址：https://blog.csdn.net/weixin_59334478/article/details/127752022