devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMq的消费确认ACK_rabbitmq channel.basicack

作者：笔触狂放9 | 2024-02-21 22:38:12

赞

踩

rabbitmq channel.basicack

1:什么是消息确认ACK

如果在处理消息的过程中，消费者的服务在处理消息的时候出现异常，那么可能这条正在处理的消息就没有完成消息消费，数据就会丢失。为了确保数据不丢失，RabbitMQ支持消息确定

2：ACK的消息确认机制

ACK机制实消费者从RabbitMQ收到消息并完成处理后，反馈给RabbitMQ，MQ收到反馈后才会将此消息从队列中删除

如果一个消费者在处理消息出现了网络不稳定服务器异常等现象，那么就不会有ACK反馈，mq会认为这个消息没有被正常消费，会将消息重新放入队列中。

如果在消息集群的情况下，mq会立即将消息推送给这个在线的其他消费者，保证了在消费者服务端故障的时候，不会丢失任何消息和任务

消息永远不会从RabbitMq中删除，只有当消费者正确发送ACK反馈，mq确认收到后才会从mq的服务器中将数据删除

3：ACK机制注意事项
如果忘记了ACK，那么后果很严重。当Consumer退出的时候，Message会一致重新分发，然后mq占据的内存越来越多，会导致内存泄漏。

4：结合项目实例

首先建立一个生产者：

  1. package com.tmnch.pinan;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import com.rabbitmq.client.MessageProperties;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import org.springframework.amqp.rabbit.support.CorrelationData;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10.  
  11. import java.io.IOException;
  12. import java.util.Date;
  13. import java.util.concurrent.TimeoutException;
  14.  
  15. /**
  16.  *  * Description: 
  17.  *  *
  18.  *  * @author thw
  19.  *  * @date 2020/7/14 14:01
  20.  *  
  21.  */
  22. public class Producer {
  23.     private static final String EXCHANGE_NAME = "exchange_pinan";
  24.     private static final String ROUTING_KEY = "routingkey_pinan";
  25.     private static final String QUEUR_NAME = "queue_pinan";
  26.     //rabbitmq的服务地址
  27.     private static final String IP_ADDRESS = "localhost";
  28.     //RabbitMq 服务端 默认端口号为5672
  29.     private static final int PORT = 5672;
  30.     //用户名
  31.     private static final String USER_NAME = "guest";
  32.     //密码
  33.     private static final String PASSWORD = "guest";
  34.  
  35.     public static void main(String[] args) throws IOException, TimeoutException {
  36.         ConnectionFactory connectionFactory = new ConnectionFactory();
  37.  
  38.         connectionFactory.setHost(IP_ADDRESS);
  39.         connectionFactory.setPort(PORT);
  40.         connectionFactory.setUsername(USER_NAME);
  41.         connectionFactory.setPassword(PASSWORD);
  42.  
  43.         /**
  44.          * 创建连接
  45.          */
  46.         Connection connection = connectionFactory.newConnection();
  47.         /**
  48.          * 创建信道
  49.          */
  50.         Channel channel = connection.createChannel();
  51.  
  52.         /**
  53.          * 创建一个type=direct 持久化的 非自动删除的交换器
  54.          * 交换器名称
  55.          * 交换器类型
  56.          * durable是否持久化
  57.          * autodelete是否自动删除
  58.          * 其他参数
  59.          */
  60.         channel.exchangeDeclare(EXCHANGE_NAME, "topic", true, false, null);
  61.  
  62.         /**
  63.          * 将交换器与队列通过路由键绑定
  64.          */
  65.         channel.queueBind(QUEUR_NAME, EXCHANGE_NAME, ROUTING_KEY);
  66.       //  channel.basicAck();
  67.    /*     channel.basicConsume(QUEUE_NAME, true, consumer);
  68.  
  69.         channel.basicConsume(QUEUR_NAME,false,)*/
  70.         /**
  71.          * 发送一条持 久化消息
  72.          */
  73.         Date date =new Date();
  74.         String message = date.toString();
  75.         message=message+"平安反渗漏接口message";
  76.         channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
  77.  
  78.         /**
  79.          * 关闭资源
  80.          */
  81.         channel.close();
  82.         connection.close();
  83.       /*  //使用rabbitmq模板
  84.         @Autowired
  85.         private RabbitTemplate rabbitTemplate;
  86. 　　　//发送消息
  87.         public void sendOrder(Order order) throws Exception{
  88. 　　　　　
  89.             CorrelationData correlationData = new CorrelationData();
  90.             correlationData.setId(order.getMessageId());
  91.  
  92.             rabbitTemplate.convertAndSend("order-exchange",//exchange
  93.                     "order.abcd",//routingKey
  94.                     order,//消息体内容
  95.                     correlationData); //消息唯一id
  96.         }*/
  97.     }
  98. }

2：新建springboot项目，引入mq依赖后，修改qpplication.yml

  1.  rabbitmq:
  2.     host: localhost  #服务器IP
  3.     port: 5672       #默认端口
  4.     username: guest  #登录名
  5.     password: guest  #密码
  6.     listener:        #消费者监听配置
  7.      simple:
  8.       acknowledge-mode: manual  #开启手动ACK确认
  9.       retry:
  10.         enabled: true           #开启重发机制
  11.         max-attempts: 5         #重发次数

新建TopicRabbitConfig.java

  1. package com.pingan.mq;
  2.  
  3. import org.springframework.context.annotation.Bean;
  4.  
  5.  
  6. import org.springframework.amqp.core.Binding;
  7. import org.springframework.amqp.core.BindingBuilder;
  8. import org.springframework.amqp.core.Queue;
  9. import org.springframework.amqp.core.TopicExchange;
  10. import org.springframework.context.annotation.Configuration;
  11. /**
  12.  *  * Description: 
  13.  *   建topic类型的mq所用
  14.  *  *
  15.  *  * @author thw
  16.  *  * @date 2020/7/29 17:38
  17.  *  
  18.  */
  19. public class TopicRabbitConfig {
  20.     //绑定键#路由键值
  21.     public final static String man = "routingkey_pinan";
  22.     @Bean
  23.     public Queue firstQueue() {
  24.         return new Queue(TopicRabbitConfig.man);
  25.     }
  26.     @Bean
  27.     TopicExchange exchange() {
  28.         return new TopicExchange("exchange_pinan");
  29.     }
  30.     //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
  31.     //这样只要是消息携带的路由键是topic.man,才会分发到该队列
  32.     @Bean
  33.     Binding bindingExchangeMessage() {
  34.         return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
  35.     }
  36. }

新建TopicReceiver.java

  1. package com.tmnch.tmnchpinan.controller;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  6. import org.springframework.amqp.support.AmqpHeaders;
  7. import org.springframework.messaging.handler.annotation.Header;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import java.io.IOException;
  11. import java.util.Map;
  12.  
  13. /**
  14.  *  * Description: 
  15.  *  *建立监听队列进入方法
  16.  *  * @author thw
  17.  *  * @date 2020/7/29 17:28
  18.  *  
  19.  */
  20. @Component
  21. @RabbitListener(queues = "queue_pinan")//监听的队列名称
  22. public class TopicReceiver {
  23.     @RabbitHandler
  24.     public void process(String testMessage, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
  25.         System.out.println("消费者消息: " + testMessage.toString());
  26.         try {
  27.             channel.basicAck(tag, false);  //第一个参数为该条消息在mq中的唯一序列号。第二个
  28. //参数为是否也确认其他消息被消费
  29.             System.out.println("序列号："+tag+":ack接收success");
  30.         } catch (IOException e) {
  31.             e.printStackTrace();
  32.             System.out.println("序列号："+tag+":ack接收fail");
  33.         }
  34.     }
  35.  
  36.  
  37. }

配送pom.xml

  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>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starter-parent</artifactId>
  8.         <version>2.3.1.RELEASE</version>
  9.         <relativePath/> <!--&lt;!&ndash; lookup parent from repository &ndash;&gt;-->
  10.     </parent>
  11.     <groupId>com.tmnch</groupId>
  12.     <artifactId>tmnchpinan</artifactId>
  13.     <version>4.0-SNAPSHOT</version>
  14.     <name>tmnchpinan</name>
  15.     <description>pingan</description>
  16.     <properties>
  17.         <java.version>1.8</java.version>
  18.     </properties>
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.springframework.boot</groupId>
  22.             <artifactId>spring-boot-starter-jdbc</artifactId>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.mybatis.spring.boot</groupId>
  26.             <artifactId>mybatis-spring-boot-starter</artifactId>
  27.             <version>1.1.1</version>
  28.         </dependency>
  29.         <dependency>
  30.             <groupId>mysql</groupId>
  31.             <artifactId>mysql-connector-java</artifactId>
  32.             <scope>runtime</scope>
  33.         </dependency>
  34.         <!--org.springframework.boot-->
  35.         <dependency>
  36.             <groupId>org.springframework.boot</groupId>
  37.             <artifactId>spring-boot-starter-test</artifactId>
  38.             <scope>test</scope>
  39.             <exclusions>
  40.                 <exclusion>
  41.                     <groupId>org.junit.vintage</groupId>
  42.                     <artifactId>junit-vintage-engine</artifactId>
  43.                 </exclusion>
  44.             </exclusions>
  45.         </dependency>
  46.         <dependency>
  47.             <groupId>org.springframework.boot</groupId>
  48.             <artifactId>spring-boot-starter-web</artifactId>
  49.             <version>1.5.4.RELEASE</version>
  50.             <!--<version>2.1.4.RELEASE</version>-->
  51.         </dependency>
  52. <!--        <dependency>
  53.             <groupId>org.springframework</groupId>
  54.             <artifactId>spring-messaging</artifactId>
  55.             <version>4.2.4.RELEASE</version>
  56.         </dependency>-->
  57.       <!--  fastjson-->
  58.         <dependency>
  59.             <groupId>com.alibaba</groupId>
  60.             <artifactId>fastjson</artifactId>
  61.             <version>1.2.6</version>
  62.         </dependency>
  63.         <!--lombok -->
  64.         <dependency>
  65.             <groupId>org.projectlombok</groupId>
  66.             <artifactId>lombok</artifactId>
  67.             <version>1.16.8</version>
  68.         </dependency>
  69.         <!--rabbitmq依赖-->
  70.         <!--<dependency>
  71.             <groupId>com.rabbitmq</groupId>
  72.             <artifactId>amqp-client</artifactId>
  73.             <version>4.1.1</version>
  74.         </dependency>
  75.         <dependency>
  76.             <groupId>org.springframework.amqp</groupId>
  77.             <artifactId>spring-rabbit</artifactId>
  78.             <version>1.6.2.RELEASE</version>
  79.         </dependency>-->
  80.         <!--rabbitmq-->
  81.         <dependency>
  82.             <groupId>org.springframework.boot</groupId>
  83.             <artifactId>spring-boot-starter-amqp</artifactId>
  84.         </dependency>
  85.         <dependency>
  86.             <groupId>org.springframework.boot</groupId>
  87.             <artifactId>spring-boot-starter</artifactId>
  88.         </dependency>
  89.  
  90.     </dependencies>
  91.  
  92.  
  93.     <!--build依赖-->
  94.     <build>
  95.         <plugins>
  96.             <plugin>
  97.                 <groupId>org.springframework.boot</groupId>
  98.                 <artifactId>spring-boot-maven-plugin</artifactId>
  99.             </plugin>
  100.         </plugins>
  101.         <resources>
  102.             <resource>
  103.                 <directory>src/main/java</directory>
  104.                 <includes>
  105.                     <include>**/*.xml</include>
  106.                 </includes>
  107.             </resource>
  108.             <resource>
  109.                 <directory>src/main/resources</directory>
  110.                 <includes>
  111.                     <include>**/*.*</include>
  112.                 </includes>
  113.             </resource>
  114.         </resources>
  115.     </build>
  116. </project>

或者

  1. # 给当前项目起名称.
  2. spring.application.name=rabbitmq-ack-direct-consumer
  3.  
  4. # 配置端口号
  5. server.port=8080
  6.  
  7. # 配置rabbitmq的参数.
  8. # rabbitmq服务器的ip地址.
  9. spring.rabbitmq.host=192.168.110.133
  10. # rabbitmq的端口号5672,区别于浏览器访问界面的15672端口号.
  11. spring.rabbitmq.port=5672
  12. # rabbitmq的账号.
  13. spring.rabbitmq.username=guest
  14. # rabbitmq的密码.
  15. spring.rabbitmq.password=guest
  16.  
  17. # 设置交换器的名称,方便修改.
  18. # 路由键是将交换器和队列进行绑定的,队列通过路由键绑定到交换器.
  19. rabbitmq.config.exchange=log.exchange.direct
  20.  
  21. # info级别的队列名称.
  22. rabbitmq.config.queue.info=log.info.queue
  23. # info的路由键.
  24. rabbitmq.config.queue.info.routing.key=log.info.routing.key
  25.  
  26. # error级别的队列名称.
  27. rabbitmq.config.queue.error=log.error.queue
  28. # error的路由键.
  29. rabbitmq.config.queue.error.routing.key=log.error.routing.key
  30.  
  31. # 开启重试
  32. spring.rabbitmq.listener.simple.retry.enabled=true
  33. # 重试次数,默认为3次
  34. spring.rabbitmq.listener.simple.retry.max-attempts=5

消费类：

  1. package com.example.bie.consumer;
  2.  
  3. import org.springframework.amqp.core.ExchangeTypes;
  4. import org.springframework.amqp.rabbit.annotation.Exchange;
  5. import org.springframework.amqp.rabbit.annotation.Queue;
  6. import org.springframework.amqp.rabbit.annotation.QueueBinding;
  7. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  9. import org.springframework.stereotype.Component;
  10.  
  11. /**
  12.  *
  13.  * @author biehl
  14.  *
  15.  *         消息接收者
  16.  *
  17.  *         1、@RabbitListener bindings:绑定队列
  18.  *
  19.  *         2、@QueueBinding
  20.  *         value:绑定队列的名称、exchange:配置交换器、key:路由键routing-key绑定队列和交换器
  21.  *
  22.  *         3、@Queue value:配置队列名称、autoDelete:是否是一个可删除的临时队列
  23.  *
  24.  *         4、@Exchange value:为交换器起个名称、type:指定具体的交换器类型
  25.  *
  26.  *
  27.  */
  28. @Component
  29. @RabbitListener(bindings = @QueueBinding(
  30.  
  31.         value = @Queue(value = "${rabbitmq.config.queue.error}", autoDelete = "true"),
  32.  
  33.         exchange = @Exchange(value = "${rabbitmq.config.exchange}", type = ExchangeTypes.DIRECT),
  34.  
  35.         key = "${rabbitmq.config.queue.error.routing.key}"))
  36. public class LogErrorConsumer {
  37.  
  38.     /**
  39.      * 接收消息的方法,采用消息队列监听机制.
  40.      *
  41.      * @RabbitHandler意思是将注解@RabbitListener配置到类上面
  42.      *
  43.      * @RabbitHandler是指定这个方法可以进行消息的接收并且消费.
  44.      *
  45.      * @param msg
  46.      */
  47.     @RabbitHandler
  48.     public void consumer(String msg) {
  49.         // 打印消息
  50.         System.out.println("ERROR消费者===>消费<===消息message: " + msg);
  51.       //  throw new RuntimeException();  用户确认ack是否正常
  52.     }
  53.  
  54. }

 

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号