RabbitMQ中的消息确认机制是什么？为什么需要消息确认？

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConfirmListener;
import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;
public class MessageConfirmationExample {
    private final static String QUEUE_NAME = "my_queue";
    private final static String MESSAGE = "Hello, RabbitMQ!";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        // 创建连接
        Connection connection = factory.newConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 启用发布确认模式
        channel.confirmSelect();
        // 创建一个有序集合，用于保存未确认的消息的Delivery Tag
        SortedSet<Long> unconfirmedSet = Collections.synchronizedSortedSet(new TreeSet<>());
        // 添加发布确认监听器
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    // 多条消息被确认，移除所有小于等于deliveryTag的消息
                    unconfirmedSet.headSet(deliveryTag + 1).clear();
                } else {
                    // 单条消息被确认，移除该消息
                    unconfirmedSet.remove(deliveryTag);
                }
                System.out.println("Message confirmed: " + deliveryTag);
            }
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    // 多条消息未被确认，重新发送所有小于等于deliveryTag的消息
                    unconfirmedSet.headSet(deliveryTag + 1).forEach(tag -> {
                        try {
                            sendMessage(channel, tag);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    });
                } else {
                    // 单条消息未被确认，重新发送该消息
                    sendMessage(channel, deliveryTag);
                }
                System.out.println("Message not confirmed: " + deliveryTag);
            }
        });
        // 发送消息
        for (int i = 1; i <= 10; i++) {
            long deliveryTag = sendMessage(channel, i);
            unconfirmedSet.add(deliveryTag);
        }
        // 等待所有消息被确认
        try {
            channel.waitForConfirmsOrDie();
            System.out.println("All messages confirmed!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 关闭通道和连接
        channel.close();
        connection.close();
    }
    private static long sendMessage(Channel channel, int messageNumber) throws IOException {
        long deliveryTag = channel.getNextPublishSeqNo();
        channel.basicPublish("", QUEUE_NAME, null, (MESSAGE + " " + messageNumber).getBytes());
        System.out.println("Sent message: " + messageNumber);
        return deliveryTag;
    }
}