springboot：整合rabbitmq之消息确认机制ack

转x33g5p2x  于2022-06-16 转载在 Spring  
字(9.3k)|赞(0)|评价(0)|浏览(637)

springboot：整合rabbitmq之消息确认机制ack

一、依赖

  1. <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-web</artifactId>
  4.         </dependency>
  6.         <dependency>
  7.             <groupId>org.springframework.boot</groupId>
  8.             <artifactId>spring-boot-starter-amqp</artifactId>
  9.         </dependency>

二、配置

  1. package com.yolo.springbootrabbitmqproducer.config;
  3. import org.springframework.amqp.core.AcknowledgeMode;
  4. import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
  5. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
  6. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import org.springframework.beans.factory.config.ConfigurableBeanFactory;
  9. import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
  10. import org.springframework.context.annotation.Bean;
  11. import org.springframework.context.annotation.Configuration;
  12. import org.springframework.context.annotation.Scope;
  14. @Configuration
  15. public class RabbitMQConfiguration {
  17.     @Bean
  18.     public ConnectionFactory rabbitConnectionFactory() {
  19.         CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
  20.         String rabbitmqHost = "127.0.0.1";
  21.         String rabbitmqPort = "5672";
  22.         String rabbitmqUsername = "guest";
  23.         String rabbitmqPassword = "guest";
  24.         String rabbitmqVirtualHost = "/";
  25.         connectionFactory.setHost(rabbitmqHost);
  26.         connectionFactory.setPort(Integer.parseInt(rabbitmqPort));
  27.         connectionFactory.setUsername(rabbitmqUsername);
  28.         connectionFactory.setPassword(rabbitmqPassword);
  29.         connectionFactory.setVirtualHost(rabbitmqVirtualHost);
  30. //        connectionFactory.setPublisherReturns(true);//开启return模式
  31. //        connectionFactory.setPublisherConfirms(true);//开启confirm模式
  32.         return connectionFactory;
  33.     }
  35.     @Bean(name = "rabbitTemplate")
  36.     //必须是prototype类型
  37.     @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  38.     public RabbitTemplate rabbitTemplate() {
  39.         return new RabbitTemplate(rabbitConnectionFactory());
  40.     }
  42.     @Bean("customContainerFactory")
  43.     public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
  44.         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  45.         //设置线程数
  46.         factory.setConcurrentConsumers(1);
  47.         //最大线程数
  48.         factory.setMaxConcurrentConsumers(1);
  49.         //设置为手动确认MANUAL(手动),AUTO(自动);
  50.         factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
  51.         // 设置prefetch
  52.         factory.setPrefetchCount(1);
  53.         configurer.configure(factory, connectionFactory);
  54.         return factory;
  55.     }
  56. }

三、消息确认机制(ack)

RabbitMQ默认的消息确认机制是：自动确认的

队列分配消息给监听消费者时，该消息处于未确认状态，不会被删除；当接收到消费者的确认回复才会将消息移除。

发送和监听消息

  1. @RestController
  2. public class AckSenderController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  7.     @GetMapping("/send")
  8.     public void send(@RequestParam(value = "message", required = false, defaultValue = "Hello World") String message) {
  9.         for (int i = 1; i <= 10; i++) {
  10.             String msg = message + " ..." + i;
  11.             System.out.println(" [ 生产者 ] Sent ==> '" + msg + "'");
  12.             rabbitTemplate.convertAndSend("helloWorldExchange","ack", msg);
  13.         }
  14.     }
  16.     private int count1=1;
  17.     private int count2=1;
  18.     @RabbitListener(containerFactory = "customContainerFactory",bindings = {@QueueBinding(
  19.             value = @Queue(value = "ackQueue",declare = "true"//指定一下队列名，默认持久队列，不指定则为临时队列
  20.             ),
  21.             exchange = @Exchange(value = "helloWorldExchange",type = ExchangeTypes.DIRECT,durable = "true"),//Exchang的默认类型就是direct，所以type可以不写
  22.             key = "ack"
  23.     )
  24.     })
  25.     public void receive(Message message) throws InterruptedException {
  26.         Thread.sleep(200);
  27.         System.out.println(" [ 消费者@1号 ] Received ==> '" + new String(message.getBody()) + "'");
  28.         System.out.println(" [ 消费者@1号 ] 处理消息数：" + count1++);
  29.     }
  31.     @RabbitListener(containerFactory = "customContainerFactory",bindings = {@QueueBinding(
  32.             value = @Queue(value = "ackQueue",declare = "true"//指定一下队列名，默认持久队列，不指定则为临时队列
  33.             ),
  34.             exchange = @Exchange(value = "helloWorldExchange",type = ExchangeTypes.DIRECT,durable = "true"),//Exchang的默认类型就是direct，所以type可以不写
  35.             key = "ack"
  36.     )
  37.     })
  38.     public void receive2(Message message) throws InterruptedException {
  39.         Thread.sleep(1000);
  40.         System.out.println(" [ 消费者@2号 ] Received ==> '" + new String(message.getBody()) + "'");
  41.         System.out.println(" [ 消费者@2号 ] 处理消息数：" + count2++);
  42.     }
  43. }

测试结果

http://localhost:8080/ack

消费者1号、2号分别拿到一条消息进行消费，但没有确认，处于阻塞状态，所以队列不会移除这两条消息，同时设置了prefetch=1，在消费者未确认之前不会重新推送消息给消费者

停止程序，发现2条未确认的消息会回到Ready里面等待重新消费

再次重启，再次消费2条消息，但仍未确认

访问/ack，再次发布消息，消息堆积

四、设置手动ack

修改消费者手动确认

  1. @RestController
  2. public class AckSenderController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  7.     private int count1=1;
  8.     private int count2=1;
  9.     private int count3 = 1;
  11.     @GetMapping("/ack")
  12.     public void send(@RequestParam(value = "message", required = false, defaultValue = "Hello World") String message) {
  13.         for (int i = 1; i <= 10; i++) {
  14.             String msg = message + " ..." + i;
  15.             System.out.println(" [ 生产者 ] Sent ==> '" + msg + "'");
  16.             rabbitTemplate.convertAndSend("helloWorldExchange","ack", msg);
  17.         }
  18.     }
  20.     @RabbitListener(containerFactory = "customContainerFactory",bindings = {@QueueBinding(
  21.             value = @Queue(value = "ackQueue",declare = "true"//指定一下队列名，默认持久队列，不指定则为临时队列
  22.             ),
  23.             exchange = @Exchange(value = "helloWorldExchange",type = ExchangeTypes.DIRECT,durable = "true"),//Exchang的默认类型就是direct，所以type可以不写
  24.             key = "ack"
  25.     )
  26.     })
  27.     public void receive(Message message, Channel channel) throws InterruptedException, IOException {
  28.         Thread.sleep(200);
  29.         System.out.println(" [ 消费者@1号 ] Received ==> '" + new String(message.getBody()) + "'");
  30.         System.out.println(" [ 消费者@1号 ] 处理消息数：" + count1++);
  31.         // 确认消息
  32.         // 第一个参数，交付标签，相当于消息ID 64位的长整数(从1开始递增)
  33.         // 第二个参数，false表示仅确认提供的交付标签；true表示批量确认所有消息(消息ID小于自身的ID)，包括提供的交付标签
  34.         channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  35.     }
  37.     @RabbitListener(containerFactory = "customContainerFactory",bindings = {@QueueBinding(
  38.             value = @Queue(value = "ackQueue",declare = "true"//指定一下队列名，默认持久队列，不指定则为临时队列
  39.             ),
  40.             exchange = @Exchange(value = "helloWorldExchange",type = ExchangeTypes.DIRECT,durable = "true"),//Exchang的默认类型就是direct，所以type可以不写
  41.             key = "ack"
  42.     )
  43.     })
  44.     public void receive2(Message message,Channel channel,@Headers Map<String, Object> map) throws InterruptedException, IOException {
  45.         Thread.sleep(600);
  46.         System.out.println(" [ 消费者@2号 ] Received ==> '" + new String(message.getBody()) + "'");
  47.         System.out.println(" [ 消费者@2号 ] 处理消息数：" + count2++);
  49.         // 确认消息
  50.         channel.basicAck((Long) map.get(AmqpHeaders.DELIVERY_TAG), false);
  51.     }
  53.     @RabbitListener(containerFactory = "customContainerFactory",bindings = {@QueueBinding(
  54.             value = @Queue(value = "ackQueue",declare = "true"//指定一下队列名，默认持久队列，不指定则为临时队列
  55.             ),
  56.             exchange = @Exchange(value = "helloWorldExchange",type = ExchangeTypes.DIRECT,durable = "true"),//Exchang的默认类型就是direct，所以type可以不写
  57.             key = "ack"
  58.     )
  59.     })
  60.     public void receive3(Message message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws InterruptedException, IOException {
  61.         Thread.sleep(1000);
  62.         System.out.println(" [ 消费者@3号 ] Received ==> '" + new String(message.getBody()) + "'");
  63.         System.out.println(" [ 消费者@3号 ] 处理消息数：" + count3++);
  65.         // 确认消息
  66.         channel.basicAck(deliveryTag, false);
  67.     }
  68. }

手动确认测试结果

手动确认通过调用方法实现
basicAck(long deliveryTag, boolean multiple)
deliveryTag：交付标签，相当于消息ID 64位的长整数(从1开始递增)
multiple：false表示仅确认提供的交付标签；true表示批量确认所有消息(消息ID小于自身的ID)，包括提供的交付标签

这里发现程序刚刚启动就全部消费完了

继续发布，还是消费完成

修改消费者手动拒绝

  1. @RestController
  2. public class AckSenderController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  7.     private int count1=1;
  8.     private int count2=1;
  9.     private int count3 = 1;
  11.     @GetMapping("/ack")
  12.     public void send(@RequestParam(value = "message", required = false, defaultValue = "Hello World") String message) {
  13.         for (int i = 1; i <= 10; i++) {
  14.             String msg = message + " ..." + i;
  15.             System.out.println(" [ 生产者 ] Sent ==> '" + msg + "'");
  16.             rabbitTemplate.convertAndSend("helloWorldExchange","ack", msg);
  17.         }
  18.     }
  20.     @RabbitListener(containerFactory = "customContainerFactory",bindings = {@QueueBinding(
  21.             value = @Queue(value = "ackQueue",declare = "true"//指定一下队列名，默认持久队列，不指定则为临时队列
  22.             ),
  23.             exchange = @Exchange(value = "helloWorldExchange",type = ExchangeTypes.DIRECT,durable = "true"),//Exchang的默认类型就是direct，所以type可以不写
  24.             key = "ack"
  25.     )
  26.     })
  27.     public void receive4(
  28.             Message message, Channel channel) throws IOException, InterruptedException {
  29.         Thread.sleep(200);
  30.         System.out.println(" [ 消费者@4号 ] Received ==> '" + new String(message.getBody()) + "'");
  31.         System.out.println(" [ 消费者@4号 ] 消息被我拒绝了：" + count1++);
  33.         // 拒绝消息方式一
  34.         // 第一个参数，交付标签
  35.         // 第二个参数，false表示仅拒绝提供的交付标签；true表示批量拒绝所有消息，包括提供的交付标签
  36.         // 第三个参数，false表示直接丢弃消息，true表示重新排队
  37.         //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
  39.         // 拒绝消息方式二
  40.         // 第一个参数，交付标签
  41.         // 第二个参数，false表示直接丢弃消息，true表示重新排队
  42.         // 跟basicNack的区别就是始终只拒绝提供的交付标签
  43.         channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
  44.     }
  45. }

手动拒绝测试结果

channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
这里是拒绝后，重新进入队列，所以消费的总是第一条消息并且循环不停
停止程序后，队列仍然是10条消息

channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
改成false，拒绝后直接丢弃
重启后：

五、总结

  • 未确认：什么也不用写，消息不会移除，重复消费，积攒越来越多
  • 确认：channel.basicAck()；确认后，消息从队列中移除
  • 拒绝：channel.basicNack()或channel.basicReject()；拒绝后，消息先从队列中移除，然后可以选择重新排队，或者直接丢弃(丢弃还有一种选择，就是加入到死信队列中，用于追踪问题)

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_43296313/article/details/125274521

内容来源于网络，如有侵权，请联系作者删除！

Javarabbitmq

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com