RabbitMQ消息可靠性（二）-- 消费者消息确认

2023-10-30

一、消费者消息确认是什么？

在这种机制下，消费者在接收到消息后，需要向 RabbitMQ 发送确认信息，告知 RabbitMQ 已经接收到该消息，并已经处理完毕。如果 RabbitMQ 没有接收到确认信息，则会将该消息重新加入队列，等待其他消费者继续处理。

消费者消息确认机制能够保证消息不会因为消费者宕机或其他原因而丢失，从而保证了消息的可靠性和稳定性。

RabbitMQ 支持两种消费者消息确认机制：自动确认和手动确认。在自动确认模式下，消费者在接收到消息后，RabbitMQ 会自动将该消息标记为已经确认。在手动确认模式下，消费者需要向 RabbitMQ 显式地发送确认信息，才能完成消息的确认。

二、代码实现

1.修改application.yml 配置

spring:
  rabbitmq:
    listener:
      simple:
        # RabbitMQ开启手动确认
        acknowledge-mode: manual

而SpringAMQP则允许配置三种确认模式：

manual：手动ack，需要在业务代码结束后，调用api发送ack。
auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack
none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

2.消费者确认

生产者发送一笔需要消费的订单到Direct Exchange直连交换机

@GetMapping("/sendDirectMessage")
    @ApiOperation(value = "sendDirectMessage")
    @ApiOperationSupport(order = 1)
    public String sendDirectMessage(@RequestParam String orderNo){
        //设置消息唯一ID
        String uniqueId = "MQ"+ DateUtils.dateTimeNow("yyyyMMddHHmmss")+ RandomUtil.randomNumbers(4);
        CorrelationData correlationData = new CorrelationData(uniqueId);
        log.info("------生产者发送消息,消息唯一id {},订单编号 {}-------",uniqueId,orderNo);
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting",orderNo,correlationData);
        return "ok";
    }

下面是消费者的处理逻辑
这里的消息序号是系统自动生成的，还需要注意的是，在手动确认模式下，如果消费者在处理消息时发生了异常或错误的时候

需要确保将该消息重新加入队列或者删除队列之后将该信息保存至数据库中记录下来，否则该消息将被认为已经成功处理并确认。因此，在编写消费者代码时，需要谨慎处理异常情况，避免因为异常而导致消息丢失或重复处理等问题。

/**
 * 消费者,用于消费队列信息
 */
@Component
@Slf4j
public class DirectConsumer {

    @Resource
    RedisService redisService;

    @RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
    public void process(Message message, Channel channel) {
        // 消息序号
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        //取出消息唯一标识
        String messageId = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        // 取出订单编码
        String orderNo = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("------消费者收到消息,消息唯一id {},订单编号 {}-------",messageId,orderNo);
        try {
            //消费者在消费消息之前，先去redis中查看消息状态是否已被消费
            if (redisService.setCacheMapIfAbsent("rabbit-tag", messageId, Boolean.FALSE)){
                //删除过期订单.......
                //消费完消息后，设置key的值为true
                redisService.setCacheMapValue("rabbit-tag", messageId, Boolean.TRUE);
                channel.basicAck(deliveryTag,false);
                log.info("------订单处理完毕,订单编号 {}--------", orderNo);
            }else {
                //如果从redis中获取消息的value是TRUE，表示已消费，直接发送确认信号，避免重复消费
                if (Boolean.TRUE.equals(redisService.getCacheMapValue("rabbit-tag",messageId))) {
                    /**
                     * TODO 手动确认消息
                     * tag：消息序号
                     * multiple：消息的标识，是否确认多条，false只确认当前一个消息收到，true确认所有consumer获得的消息（成功消费，消息从队列中删除
                     */
                    channel.basicAck(deliveryTag, false);
                    log.info("--------订单已经被消费过了,订单编号 {}-------", orderNo);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            try {
                /**
                 * TODO 消费者消费消息异常，手动否认信息，将消息退回到队列中
                 * tag：消息序号
                 * multiple：消息的标识，是否确认多条，false只确认当前一个消息收到，true确认所有consumer获得的消息（成功消费，消息从队列中删除
                 * requeue：是否要退回到队列
                 */
                channel.basicNack(deliveryTag, true, false);
                redisService.setCacheMapValue("rabbit-tag", messageId, Boolean.FALSE);
                log.error("------------订单消费失败,已从队列删除.订单编号 {}, 原因 {}--------",orderNo, e.getMessage());
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
        log.info("------消费者处理完毕-------");
    }
}

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)