RabbitMQ如何保证消息不丢失

未分类 2年前 (2022) 程序员胖胖胖虎阿

186 0 0

RabbitMQ如何保证消息不丢失

消息传输中的三个过程

生产端到RabbitMQ、RabbitMQ到消费端、消费端消费掉消息

在这三个过程中，任意一过程都将可能导致消息传输处理失败

生产端到RabbitMQ

事务消息机制：会严重降低性能
confirm消息确认机制：生产端投递消息一旦投递到RabbitMQ后，RabbitMQ就会发送一个确认消息给生产端

开启消息确认的方式：

channel.confirmSelect();// 开启发送方确认模式

异步监听确认和未确认的消息

channel.addConfirmListener(new ConfirmListener() {
    //消息正确到达broker
    @Override
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("已收到消息");
        //做一些其他处理
    }
    //RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息
    @Override
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("未确认消息，标识：" + deliveryTag);
        //做一些其他处理，比如消息重发等
    }
});

此时，无论RabbitMQ是否确认消息，生产端都可以感知到消息是否处理成功。

问题：但是如果RabbitMQ接收到消息，并进行处理，但还未来的及发送确认消息，或者生产端由于网络故障没有接收到消息，，生产端就不知道消息有没有发送成功？

消息持久化

消息持久化解决的是RabbitMQ突然挂掉了导致内存中消息丢失的情况。为了让重启后的RabbitMQ 重启后，能够恢复硬盘中的数据，我们需要将exchange、queue和message都进行持久化

//exchange持久化
//第三个参数true表示这个exchange持久化
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);

//queue持久化
//第二个参数true表示这个queue持久化
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

//message 持久化
//第三个参数MessageProperties.PERSISTENT_TEXT_PLAIN表示这条消息持久化
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));

问题：RabbitMQ接收到消息，还未来得及持久化到硬盘的时候，RabbitMQ挂了，消息丢失

* 消息入库

为了保证消息传输过程中，消息不被丢失，将发送消息前，先将消息保存到数据库，增加status状态字段、超时时间以及重发次数。其中状态字段表示当时消息传输是否传输成功。超时时间表示当超过一定时间后，重新发送消息。（可以通过定时任务，定时检测status为0，且时间大于超时时间的消息），重发次数表示最大的重发次数，如果发送一直失败，超过一定次数，就对做单独处理

问题：当目前为止，还存在一个问题，如果消息到了RabbitMQ，但是确认消息接收失败，这种，如果通过重复发送消息的方式可能导致消息的冗余，目前可想到的方式但不一定是最优的方案是，在消息体中，增加一个唯一ID，确定消息的唯一性。

消费端消息不丢失

消费端的话，可以认为有两种情况，消息传输过程中和消息处理过程中，导致消息处理失败。

如果消费端和生产端一样，通过ack机制确认，那么这种问题可以避免。默认RabbitMQ在消息发出后就立即删除这条消息，导致无论消费端是否接受到消息，是否处理完，消息都会丢失

所以，需要将自动ack机制改为手动ack机制

消费端手动确认消息

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    try {
        //接收到消息，做处理
        //手动确认
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        //出错处理，这里可以让消息重回队列重新发送或直接丢弃消息
    }
};
//第二个参数autoAck设为false表示关闭自动确认机制，需手动确认
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});