• 十七、SpringAMQP

    目录

    一、SpringAMQP的介绍：

    二、利用SpringAMQP实现HelloWorld中的基础消息队列功能

    1、因为publisher和consumer服务都需要amqp依赖，因此这里把依赖直接放到父工程mq-demo中

    2、编写yml文件

    3、编写测试类，并进行测试

    三、在consumer中编写消费逻辑，监听simple.queue

    1、导入依赖，刚才在父工程中已经导入了，所以省略

    2、编写yml文件

    3、新建类，实现消费逻辑

    4、运行并测试

    四、模拟WorkQueue，实现一个队列绑定多个消费者

    1、编写生产者（生产50个消息）

    2、编写消费者（一个消费者更快，一个消费者更慢）

    3、测试

    4、消费预取的修改

    5、重新测试

    五、发布和订阅

    （一）利用SpringAMQP演示FanoutExchange的使用

    1、新建config类，声明交换机和队列

    2、启动项目，查看配置

    3、编写消费者代码

    4、编写生产者代码

    5、运行代码，观察输出

    （二）交换机的作用

    （三）声明队列、交换机、绑定关系的Bean是什么?

    （四）DirectExchange

    1、编写消费者代码

    2、编写生产者代码

    （五）Direct交换机与Fanout交换机的差异

    （六）TopicExchange

    1、编写消费者代码

    2、编写生产者代码

    3、运行测试

    4、描述下Direct交换机与Topic交换机的差异

    （七）测试发送Object类型信息

    1、新增队列

    2、发送对象

    3、查看

    4、优化（使用jackson进行序列化）

    5、接收消息

    一、SpringAMQP的介绍：

    1. AMQP是一种高级消息队列协议。

    2. SpringAMQP是基于Spring Framework的AMQP扩展，提供了一个抽象层，使得使用AMQP进行消息传递变得更加简单。

    3. SpringAMQP支持多种消息传递模式，包括点对点、发布/订阅和请求/响应等。

    4. SpringAMQP提供了许多高级功能，例如队列管理、消息确认、事务和消息过滤等。

    5. SpringAMQP提供了集成测试工具和基于Spring Boot的自动配置，使得集成AMQP变得更加容易。

    6. 总之，SpringAMQP是一个灵活、可扩展的AMQP实现，它使得使用消息队列时变得更加容易和高效。

    二、利用SpringAMQP实现HelloWorld中的基础消息队列功能

    1、因为publisher和consumer服务都需要amqp依赖，因此这里把依赖直接放到父工程mq-demo中

    1. <!--        AMQP依赖，包含RabbitMQ-->
    2.         <dependency>
    3.             <groupId>org.springframework.boot</groupId>
    4.             <artifactId>spring-boot-starter-amqp</artifactId>
    5.         </dependency>

    2、编写yml文件

    1. logging:
    2.   pattern:
    3.     dateformat: MM-dd HH:mm:ss:SSS
    4. spring:
    5.   rabbitmq:
    6.     host: 192.168.248.152
    7.     port: 5672
    8.     virtual-host: /
    9.     username: itcast
    10.     password: 123456
    11.

    3、编写测试类，并进行测试

    1. @RunWith(SpringRunner.class)
    2. @SpringBootTest
    3. public class SpringAmqpTest {
    4.     @Autowired
    5.     private RabbitTemplate rabbitTemplate;
    6.  
    7.     @Test
    8.     public void testSendMassage(){
    9.         String queue = "simple.queue";
    10.         String massage="aaaaaaa";
    11.         rabbitTemplate.convertAndSend(queue,massage);
    12.     }
    13. }

    三、在consumer中编写消费逻辑，监听simple.queue

    1、导入依赖，刚才在父工程中已经导入了，所以省略

    2、编写yml文件

    1. logging:
    2.   pattern:
    3.     dateformat: MM-dd HH:mm:ss:SSS
    4. spring:
    5.   rabbitmq:
    6.     host: 192.168.248.152
    7.     port: 5672
    8.     virtual-host: /
    9.     username: itcast
    10.     password: 123456
    11.

    3、新建类，实现消费逻辑

    1. package cn.itcast.mq.listener;
    2.  
    3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    4. import org.springframework.stereotype.Component;
    5.  
    6. @Component
    7. public class SpringRabbitListener {
    8.     @RabbitListener(queues = "simple.queue")
    9.     public void listenSimpleQueue(String msg){
    10.         System.out.println("消费者接收到消息："+msg);
    11.     }
    12. }

    4、运行并测试

    注意:

    消息一旦消费就会从队列删除，RabbitMQ没有消息回溯功能

    四、模拟WorkQueue，实现一个队列绑定多个消费者

    1、编写生产者（生产50个消息）

    1. @RunWith(SpringRunner.class)
    2. @SpringBootTest
    3. public class SpringAmqpTest {
    4.     @Autowired
    5.     private RabbitTemplate rabbitTemplate;
    6.  
    7.     @Test
    8.     public void testSendMassage(){
    9.         String queue = "simple.queue";
    10.         String massage="HelloWorld";
    11.  
    12.         for (int i = 0; i < 50; i++) {
    13.             rabbitTemplate.convertAndSend(queue,massage);
    14.         }
    15.     }
    16. }

    2、编写消费者（一个消费者更快，一个消费者更慢）

    1. @Component
    2. public class SpringRabbitListener {
    3.     @RabbitListener(queues = "simple.queue")
    4.     public void listenSimpleQueue(String msg){
    5.         System.out.println("消费者0接收到消息："+msg+ LocalTime.now());
    6.         try {
    7.             Thread.sleep(20);
    8.         } catch (InterruptedException e) {
    9.             throw new RuntimeException(e);
    10.         }
    11.     }
    12.  
    13.     @RabbitListener(queues = "simple.queue")
    14.     public void listenSimpleQueue1(String msg){
    15.         System.err.println("消费者1接收到消息__________-："+msg+ LocalTime.now());
    16.         try {
    17.             Thread.sleep(200);
    18.         } catch (InterruptedException e) {
    19.             throw new RuntimeException(e);
    20.         }
    21.     }
    22. }

    3、测试

    我们发现，虽然消费者0更快，但是它并没有承担更多的工作量；

    这是因为消费预取机制会让消费者事先分配好要处理的消息，而不是按能力分配；

    4、消费预取的修改

    可以在yml文件中修改

    1.     listener:
    2.       simple:
    3.         prefetch: 1 #表示预取上限为1

    5、重新测试

    五、发布和订阅

    （一）利用SpringAMQP演示FanoutExchange的使用

    1、新建config类，声明交换机和队列
    1. @Configuration
    2. public class FanoutConfig {
    3.     ///1
    4.     @Bean
    5.     public FanoutExchange fanoutExchange(){
    6.         return new FanoutExchange("itcast.fanout");
    7.     }
    8.  
    9.     @Bean
    10.     public Queue fanoutQueue1(){
    11.         return new Queue("fanout.queue1");
    12.     }
    13.  
    14.     @Bean
    15.     public Binding bindingQueue1(FanoutExchange exchange,Queue fanoutQueue1){
    16.         return BindingBuilder.bind(fanoutQueue1).to(exchange);
    17.     }
    18.  
    19.     ///2
    20.     @Bean
    21.     public Queue fanoutQueue2(){
    22.         return new Queue("fanout.queue2");
    23.     }
    24.  
    25.     @Bean
    26.     public Binding bindingQueue2(FanoutExchange exchange,Queue fanoutQueue2){
    27.         return BindingBuilder.bind(fanoutQueue2).to(exchange);
    28.     }
    29. }
    2、启动项目，查看配置

    绑定成功

    3、编写消费者代码
    1.     @RabbitListener(queues = "fanout.queue1")
    2.     public void listenFanoutQueue1(String msg){
    3.         System.err.println("消费者1接收到消息__________-："+msg+ LocalTime.now());
    4.     }
    5.  
    6.     @RabbitListener(queues = "fanout.queue2")
    7.     public void listenFanoutQueue2(String msg){
    8.         System.err.println("消费者2接收到消息__________-："+msg+ LocalTime.now());
    9.     }
    4、编写生产者代码
    1.     @Test
    2.     public void sendFanoutMassage(){
    3.         String exchangeName = "itcast.fanout";
    4.         String message = "Hello Every One";
    5.         rabbitTemplate.convertAndSend(exchangeName,"",message);
    6.     }
    5、运行代码，观察输出

    发现两个消费者都接收到了消息

    （二）交换机的作用

    • 接收publisher发送的消息
    • 将消息按照规则路由到与之绑定的队列
    • 不能缓存消息，路由失败，消息丢失
    • FanoutExchange的会将消息路由到每个绑定的队列

    （三）声明队列、交换机、绑定关系的Bean是什么?

    • Queue
    • FanoutExchange
    • Binding
       

    （四）DirectExchange

    实现：

    1、编写消费者代码
    1.     @RabbitListener(bindings = @QueueBinding(
    2.             value = @Queue(name = "direct.queue2"),
    3.             exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
    4.             key = {"red","blue"}
    5.     ))
    6.     public void listenDirectQueue1(String msg){
    7.         System.err.println("消费者1接收到消息__________-："+msg+ LocalTime.now());
    8.     }
    9.  
    10.     @RabbitListener(bindings = @QueueBinding(
    11.             value = @Queue(name = "direct.queue2"),
    12.             exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
    13.             key = {"red","yellow"}
    14.     ))
    15.     public void listenDirectQueue2(String msg){
    16.         System.out.println("消费者2接收到消息__________-："+msg+ LocalTime.now());
    17.     }
    2、编写生产者代码
    1.     @Test
    2.     public void sendDirectMassage(){
    3.         String exchangeName = "itcast.direct";
    4.         String message = "Hello Every One1111";
    5.         rabbitTemplate.convertAndSend(exchangeName,"blue",message);
    6.     }
    1.     @Test
    2.     public void sendDirectMassage(){
    3.         String exchangeName = "itcast.direct";
    4.         String message = "Hello Every One1111";
    5.         rabbitTemplate.convertAndSend(exchangeName,"red",message);
    6.     }

    （五）Direct交换机与Fanout交换机的差异

    • Fanout交换机将消息路由给每一个与之绑定的队列
    • Direct交换机根据RoutingKey判断路由给哪个队列
    • 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
       

    基于@RabbitListener注解声明队列和交换机有哪些常见注解

    • @Queue
    • @Exchange
       

    （六）TopicExchange

    利用SpringAMQP演示TopicExchange的使用

    1、编写消费者代码
    1.     @RabbitListener(bindings = @QueueBinding(
    2.             value = @Queue(name = "topic.queue1"),
    3.             exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
    4.             key = "china.#"
    5.     ))
    6.     public void listenTopicQueue1(String msg){
    7.         System.out.println("消费者1接收到消息aaaaaa__-："+msg+ LocalTime.now());
    8.     }
    9.  
    10.     @RabbitListener(bindings = @QueueBinding(
    11.             value = @Queue(name = "topic.queue2"),
    12.             exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
    13.             key = "#.news"
    14.     ))
    15.     public void listenTopicQueue2(String msg){
    16.         System.err.println("消费者2接收到消息a__-："+msg+ LocalTime.now());
    17.     }
    2、编写生产者代码
    1.     @Test
    2.     public void sendTopicMassage(){
    3.         String exchangeName = "itcast.topic";
    4.         String message = "Hello Every One12222";
    5.         rabbitTemplate.convertAndSend(exchangeName,"china.news",message);
    6.     }
    3、运行测试

    4、描述下Direct交换机与Topic交换机的差异

    （七）测试发送Object类型信息

    1、新增队列
    1.     @Bean
    2.     public Queue objectQueue(){
    3.         return new Queue("object.queue");
    4.     }
    2、发送对象
    1.     @Test
    2.     public void sendObjectMassage(){
    3.         Map message = new HashMap<>();
    4.         message.put("name","11");
    5.         message.put("age","22");
    6.         rabbitTemplate.convertAndSend("object.queue",message);
    7.     }
    3、查看

    对象被序列化了，这种方式性能差，不安全（容易被注入）

    4、优化（使用jackson进行序列化）

    引入依赖 

    1.         
    2.             com.fasterxml.jackson.core
    3.             jackson-databind
    4.

    添加配置Bean

    1.     @Bean
    2.     public Jackson2JsonMessageConverter messageConverter(){
    3.         return new Jackson2JsonMessageConverter();
    4.     }

    5、接收消息

    编写配置Bean

    1.     @Bean
    2.     public Jackson2JsonMessageConverter messageConverter(){
    3.         return new Jackson2JsonMessageConverter();
    4.     }

    编写消费者代码

    1. @RabbitListener(queues = "object.queue")
    2. public void listenObjectQueue1(Map msg){
    3.     System.err.println("消费者接收到消息___da_______-："+msg+ LocalTime.now());
    4. }

    验证

    注意：

  • 相关阅读:
    基础I/O【Linux】
    09【SpringMVC的Json支持】
    unity urp 实现衣服绒毛效果
    DVWA安装教程（懂你的不懂·详细）
    kettle从入门到精通 第五十三课 ETL之kettle MQTT/RabbitMQ consumer实战
    Java Story
    【9】Docker的迁移与备份
    Lua01 基本语法+数据类型+流程控制+运算符
    【微信小程序】获取用户手机号的实现
    Android Surface 是如何跨进程传递的？
  • 原文地址：https://blog.csdn.net/icbbm/article/details/134530894