梦应逍遥 发表于 2024-7-29 09:36:37

RabbitMq手动ack的超简单案例+Confirm和Return机制的配置和利用

最简单的例子

先简单介绍一下这三个方法
basicAck

表现确认乐成,利用此方法后,消息会被rabbitmq broker删除
https://img-blog.csdnimg.cn/img_convert/27da0e17a96540444c8c02bfe3805405.png

https://img-blog.csdnimg.cn/img_convert/8da61018bebe0b359ce601d31c44d14f.png
 
basicNack

表现失败确认,一样平常在消费消息业务异常时用到此方法,可以将消息重新投递入队列

https://img-blog.csdnimg.cn/img_convert/ebbe29b0e3fe435563428b32a7db686d.png

https://img-blog.csdnimg.cn/img_convert/26a1dc691a905862849c730d29014aca.png
basicReject

拒绝消息,与basickNack区别在于不能举行批量操纵,其他用法很相似

https://img-blog.csdnimg.cn/img_convert/331d70485b18dddf495f3e74a17b77fd.png
形参

multiple表现是否批量处理
requeue表现是否重复入队
deliverTag:表现消息投递序号,每次消费消息或者消息重新投递后,deliverTag都会增长。手动消息确认模式下,我们可以对指定deliverTag的消息举行ack、nack、reject等操纵。
mutiple:是否批量确认,值为 true 则会一次性 ack全部小于当前消息 delivertag 的消息
依赖

<dependencies>
      <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
      </dependency>

      <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
      </dependency>
      <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-amqp</artifactId>
      </dependency>
      <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
      </dependency>
      <dependency>
            <groupId>com.alibaba.fastjson2</groupId>
            <artifactId>fastjson2</artifactId>
            <version>2.0.51</version>
      </dependency>
      <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
      </dependency>
      <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
      </dependency>
    </dependencies> springboot配置

spring:
rabbitmq:
    host: 192.168.88.130
    port: 5672
    username: itcast
    password: 123321
    virtual-host: /   #虚拟主机,默认是/,RabbitMQ 使用虚拟主机来隔离不同的消息环境,虚拟主机用于将消息、交换器、队列和绑定隔离开来
    publisher-confirm-type: correlated #发布者消息确认功能(异步)
    listener:
      simple:
      retry:
          enabled: true#开启消费者失败重试
          max-attempts: 5 #最大重试次数
          initial-interval: 1000ms #初始失败等待时长为1秒
          multiplier: 1 #下次失败的等待时长倍数,下次等待时长=multipiler*last-interval
      acknowledge-mode: manual #开启手动ack机制 auto是自动 none是发送后直接ack(这个不会用上的)
    publisher-returns: true #发布者返回消息功能

logging:
level:
    com.atguigu.mq.config.MQProducerAckConfig: info 交换机和队列配置

package com.example.rabbitmq.Configuration;


import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.ImmediateRequeueMessageRecoverer;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class QueueConfig {

    @Bean(name = "confirmTestQueue")
    public Queue confirmTestQueue() {
      return new Queue("confirm_test_queue", true, false, false);
    }

    @Bean(name = "confirmTestExchange")
    public FanoutExchange confirmTestExchange() {
      return new FanoutExchange("confirmTestExchange");
    }

    @Bean
    public Binding confirmTestFanoutExchangeAndQueue(
            @Qualifier("confirmTestExchange") FanoutExchange confirmTestExchange,
            @Qualifier("confirmTestQueue") Queue confirmTestQueue) {
      return BindingBuilder.bind(confirmTestQueue).to(confirmTestExchange);
    }



}
消费者

我设置了最多重试三次
有一个小细节
就是他重试的时候是在队头重试的
所以他重试的时候会阻塞一段时间,此时后面的消息是不能消费的
package com.example.rabbitmq.Listener;


import com.alibaba.fastjson2.JSON;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.tools.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Slf4j
@Component
public class ReceiverMessage1 {

    //用来存放消息唯一标识的map,设置一定的重试次数
    public static final ConcurrentMap<String, Integer> map = new ConcurrentHashMap<>();

    @RabbitListener(queues = "confirm_test_queue")
    public void getMessage3(String msg, Channel channel, Message message) throws IOException {


      //得到当前信息的唯一标识
      String messageId=message.getMessageProperties().getMessageId();

      try {
            System.out.println("成功接收到消息:" + msg);

       int i=1/0;
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
log.info("确认成功");
      } catch (Exception e) {



            map.put(messageId,map.getOrDefault(messageId, 0)+1);

            log.error("接收消息失败");
          log.info("开始重试");
          log.info(messageId);

          //重复处理失败
          if(map.get(messageId)<=3) {
            log.info("确认失败,重新入队");
             channel.basicNack(message.getMessageProperties().getDeliveryTag(),false, true);
   map.put(messageId, map.getOrDefault(messageId, 0)+1);

          }
else {

    log.info("重试仍然失败,所以我们决定丢弃这个消息");

            channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);

          }


      }


    }



    }
生产者(测试类)

package com.example.rabbitmq;

import org.apache.catalina.Executor;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.doThrow;

@SpringBootTest
class RabbitmqApplicationTests {

    @Test
    void contextLoads() {
    }

    @Autowired
    private RabbitTemplate rabbitTemplate;



    @Test
    void sendMessage() {
       rabbitTemplate.convertAndSend("confirmTestExchange",
               "confirm_test_queue", "这是一个测试消息",message -> {
         message.getMessageProperties().setMessageId(UUID.randomUUID().toString());//把消息的唯一标识设置为UUID
         return message;
               });
    }



}
修改了messageId

因为messageId是基于交换机,内容,队列来生成的
相同的消息可能messageId是一样的
所以我发送消息的时候把底层改成了UUID

https://img-blog.csdnimg.cn/img_convert/61e1277187b2dfd2ad7219c9cf1dc773.png

Return和Confirm机制


https://img-blog.csdnimg.cn/img_convert/4721293cf75cd8bc8ff54d68d84d8955.png

https://img-blog.csdnimg.cn/img_convert/2ba0c55a3bcb6ff80caf11985301aaaa.png
Confirm机制是消息发送到交换机乐成或失败时的处理机制

https://img-blog.csdnimg.cn/img_convert/e789a4d7a8a6c39afef00045968a677b.png
Retrun机制是消息发送到队列失败时的处理机制
https://img-blog.csdnimg.cn/img_convert/086309d922fdece7ab9d8507e3e68756.png
配置(加了日志输出)

package com.example.rabbitmq.Configuration;

import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class RabbitConfig implements RabbitTemplate.ReturnsCallback,RabbitTemplate.ConfirmCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

//在Bean注入前就要执行的方法
    @PostConstruct
    privatevoid initRabbitTemplate(){

      rabbitTemplate.setReturnsCallback(this);
      rabbitTemplate.setConfirmCallback(this);

    }


    //消息发送到交换机成功或失败时调用这个方法
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
log.info("confirm()回调函数大打印correlationData:"+correlationData);
log.info("confirm()回调函数大打印ack:"+ack);
log.info("confirm()回调函数大打印cause:"+cause);

    }

    //发送到队列失败的时候调用这个方法
    @Override
    public void returnedMessage(ReturnedMessage returned) {
      log.info("消息主体: " + new String(returned.getMessage().getBody()));
      log.info("应答码: " + returned.getReplyCode());
      log.info("描述:" + returned.getReplyText());
      log.info("消息使用的交换器 exchange : " + returned.getExchange());
      log.info("消息使用的路由键 routing : " + returned.getRoutingKey());

    }
}
遇到的小问题


它会有个报错
2024-07-28T14:00:10.866+08:00 ERROR 45104 --- o.s.a.r.c.CachingConnectionFactory : Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
这个是因为我们的多次ack,导致的错误
例如我们可能多次ac或者reject了,就会发生这种错误
这是因为我们的配置默认是自动ack
假如我们不开启手动ack,我们的本身写的手动ack代码就算是再次ack了
所以会出现这个通道错误
这个会让我们的mq通道断连,然后再重连,这样子会导致部门消息丢失,所以记得配置开启手动ack
https://img-blog.csdnimg.cn/img_convert/eeeeb29ed23471edba1190907fb1a80a.png
发送消息回队尾

这个我处理失败了,我也不知道为什么,可能我本身修改了messageId吧
反正这个重新发送消息,我得到的和之前的不同,有部门属性缺失了
所以我就没弄这个
假如想具体相识发起看参考文章

https://img-blog.csdnimg.cn/img_convert/44255e694d74a38111225c2be01579d8.png
参考文章
Springboot + RabbitMQ 用了消息确认机制,感觉掉坑里了!-腾讯云开辟者社区-腾讯云 (tencent.com)











免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。
页: [1]
查看完整版本: RabbitMq手动ack的超简单案例+Confirm和Return机制的配置和利用