【RabbitMQ-8】消息确定机制（自动ACK和手动ACK）

1. 消息确定机制及其配置

RabbitMq消费者的消息确定机制：

  • NONE：无应答，rabbitmq默认consumer正确处理所有请求。
  • AUTO：consumer自动应答，处理成功（注意：此处的成功确认是没有发生异常）发出ack，处理失败发出nack。rabbitmq发出消息后会等待consumer端应答，只有收到ack确定信息后才会将消息在rabbitmq清除掉。收到nack异常信息的处理方法由setDefaultRequeueReject()方法设置，这种模式下，发送错误的消息可以恢复。
  • MANUAL：基本等同于AUTO模式，区别是需要人为调用方法确认。

在配置文件中：

spring:
  profiles:
    active: test
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    template:
      receive-timeout: 2000
      reply-timeout: 2000
    listener:
      simple:
        concurrency: 1
        max-concurrency: 3
        # 消费者预取1条数据到内存，默认为250条
        prefetch: 1
        # 确定机制
        acknowledge-mode: manual

可以进行配置，也可以在代码中进行配置：

@Configuration
public class RabbitConfig {

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        /* setConnectionFactory：设置spring-amqp的ConnectionFactory。 */
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(10);
        factory.setPrefetchCount(1);
        factory.setDefaultRequeueRejected(true);
        //手动确认。
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }
}

2. 自动确认和手动确认细节

2.1 测试代码配置

队列配置：

@Configuration
public class RabbitMqConfig {
    // 简单的声明一个队列
    @Bean
    public Queue kinsonQueue1() {
        return new Queue("kinson1");
    }
}

生产者配置：

@RestController
public class RabbitMQController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    //直接向队列中发送数据
    @GetMapping("send")
    public String send() {
        for (int i = 0; i < 10; i++) {
            String content = "Date:" + System.currentTimeMillis();
            content = content + ":::" + i;
            rabbitTemplate.convertAndSend("kinson1",i+"");
        }
        return "success";
    }
}

消费者配置

@Component
@Slf4j
public class CustomerRev {

    //消费者处理消息缓慢
    @RabbitListener(queues = {"kinson1"})
    public void receiver3(Message msg, Channel channel) {
        try {
            //打印数据
            String message = new String(msg.getBody(), StandardCharsets.UTF_8);
            log.info("【开始】：{}",message);
            Thread.sleep(30000);
            log.info("【结束】:{}", message);
        } catch (Exception e) {
            log.error("错误信息：{}", e.getMessage());
        }
    }
}

2.2 自动确认

生产者产生10笔消息，自动确认模式下，消息处理成功，消费者才会去获取下一笔消息；消息处理抛出异常，那么将会消息重回队列。

自动确认.png

注：此时的prefetch=1，该消费者unack的消息只有一条。其余9条均在MQ中。

若是设置prefetch=250（默认），那么消费者会预期10条消息，unack的消息是10条，但是日志中依旧只是消费了一条消息。

【开始】：Date:1604557983496:::0
prefetch为250的情况.png

总结：对于自动确认来说，当方法没有异常执行完毕后，会对MQ发出ACK；若方法出现异常，会对MQ发出nack，消息会重回队列。要分清哪些是可以恢复的异常，哪些是不可以恢复的异常。不可恢复的异常，在消费者代码中捕获异常，并记录日志表或放入死信队列。可恢复的异常，那么放入业务队列中重试。

2.3 手动确认

手动确认常用的API：

//ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);

//Reject表示拒绝消息。requeue：false表示被拒绝的消息是丢弃；true表示重回队列
channel.basicReject(msg.getMessageProperties().getDeliveryTag(),false);

 //nack表示拒绝消息。multiple表示拒绝指定了delivery_tag的所有未确认的消息，requeue表示不是重回队列
channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,false);

案例：消费者只有发送ACK后才会开始消费下一条消息。

@Component
@Slf4j
public class CustomerRev {

    //消费者处理消息缓慢
    @RabbitListener(queues = {"kinson1"})
    public void receiver3(Message msg, Channel channel) {
        try {
            //打印数据
            String message = new String(msg.getBody(), StandardCharsets.UTF_8);
            log.info("【开始】：{}",message);
            if("0".equals(message)){
                Thread.sleep(1000);
            }else {
                Thread.sleep(300000000L);
            }
            log.info("【结束】:{}", message);
            //ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            log.error("错误信息：{}", e.getMessage());
        }
    }
}

总结：手动确认模式下，返回队列便是进行重试。

3. 消息重回队列

自动确认模式下抛出异常，和手动确认模式下nack消息都可以重回队列。

3.1 消息nack后重回队头

@Component
@Slf4j
public class CustomerRev {

    //消费者处理消息缓慢
    @RabbitListener(queues = {"kinson1"})
    public void receiver3(Message msg, Channel channel) {
        try {
            //打印数据
            String message = new String(msg.getBody(), StandardCharsets.UTF_8);
            log.info("【开始】：{}",message);
            if("0".equals(message)){
                throw new RuntimeException("0的消息消费异常");
            }
            log.info("【结束】:{}", message);
            //ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

（自动ACK下）当消息为0时，抛出异常，生产者产生10笔消息。

image.png

代码轮询打印0的消息的消费异常，其他消息均不消费。


image.png

当消息回滚到消息队列时，这条消息不会回到队列尾部，而是仍是在队列头部，这时消费者会立马又接收到这条消息进行处理，接着抛出异常，进行 回滚，如此反复进行。这种情况会导致消息队列处理出现阻塞，消息堆积，导致正常消息也无法运行。

3.2 消息重发送到队尾

    //消费者处理消息缓慢
    @RabbitListener(queues = {"kinson1"})
    public void receiver3(Message msg, Channel channel) throws IOException {
        try {
            //打印数据
            String message = new String(msg.getBody(), StandardCharsets.UTF_8);
            log.info("【开始】：{}",message);
            if("0".equals(message)){
                throw new RuntimeException("0的消息消费异常");
            }
            log.info("【结束】:{}", message);
            //ack表示确认消息。multiple：false只确认该delivery_tag的消息，true确认该delivery_tag的所有消息
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            //捕获异常后，重新发送到指定队列，自动ack不抛出异常即为ack
            channel.basicPublish(msg.getMessageProperties().getReceivedExchange(),
                    msg.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
                    msg.getBody());
        }
    }

手动确认的模式：

//手动进行应答
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//重新发送消息到队尾
channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
JSON.toJSONBytes(new Object()));
最终mq剩余的消息.png

这个不会导致其他消息被阻塞消费，但是会大量的抛出：

错误日志.png

3.3 解决不可处理异常

如果一个消息体本身有误，会导致该消息体，一直无法进行处理，而服务器中刷出大量无用日志。解决这个问题可以采取两种方案：

1.一种是对于日常细致处理，分清哪些是可以恢复的异常，哪些是不可以恢复的异常。对于可以恢复的异常我们采取第三条中的解决方案，对于不可以处理的异常，我们采用记录日志，直接丢弃该消息方案。

2.另一种是我们对每条消息进行标记，记录每条消息的处理次数，当一条消息，多次处理仍不能成功时，处理次数到达我们设置的值时，我们就丢弃该消息，但需要记录详细的日志。

将业务队列绑定死信队列，当消息被丢弃后，进入到死信队列（代码修复后监听死信队列补偿消息）。可以避免我们手动的恢复消息。

重试机制（自动确认）：

@Component
@Slf4j
public class CustomerRev {

    @RabbitListener(queues = {"kinson1"})
    public void receiver3(Message msg, Channel channel) throws IOException {
        try {
            //打印数据
            String message = new String(msg.getBody(), StandardCharsets.UTF_8);
            log.info("【开始】：{}",message);
            if("0".equals(message)){
                throw new RuntimeException("0的消息消费异常");
            }
            log.info("【结束】:{}", message);
        } catch (Exception e) {
            //捕获异常后，重新发送到指定队列，自动确认不抛出异常即为ack
            Integer retryCount;
            Map<String, Object> headers = msg.getMessageProperties().getHeaders();
            if(!headers.containsKey("retry-count")){
                retryCount=0;
            }else {
                retryCount = (Integer)headers.get("retry-count");
            }
            //判断是否满足最大重试次数(重试3次)
            if(retryCount++<3) {
                headers.put("retry-count",retryCount);
                //重新发送到MQ中
                AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().contentType("text/plain").headers(headers).build();
                channel.basicPublish(msg.getMessageProperties().getReceivedExchange(),
                        msg.getMessageProperties().getReceivedRoutingKey(), basicProperties,
                        msg.getBody());
            }
        }
    }
}

每次收到消息后，修改消息的header信息。来记录重试的次数。

推荐阅读

rabbitmq消息重回队列

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 214,875评论 6赞 496
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 91,569评论 3赞 389
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 160,475评论 0赞 350
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 57,459评论 1赞 288
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 66,537评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,563评论 1赞 293
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,580评论 3赞 414
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,326评论 0赞 270
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,773评论 1赞 307
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,086评论 2赞 330
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,252评论 1赞 343
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,921评论 5赞 338
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,566评论 3赞 322
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,190评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,435评论 1赞 268
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 47,129评论 2赞 366
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,125评论 2赞 352