// 偽代碼示例：生產(chǎn)端確認(rèn)模式
public void sendWithConfirm(Message msg) {
    // 1. 持久化到本地?cái)?shù)據(jù)庫(kù)（防丟失）
    messageDao.save(msg);
    // 2. 發(fā)送到消息隊(duì)列
    String msgId = rabbitTemplate.convertAndSend(msg);
    // 3. 等待Broker確認(rèn)
    boolean ack = waitForAck(msgId, TIMEOUT);
    // 4. 失敗重試（指數(shù)退避）
    if (!ack) {
        retryWithBackoff(msg);
    }
    // 5. 最終記錄投遞狀態(tài)
    updateDeliveryStatus(msgId, ack);
}

消息處理流程：
Producer → Broker接收 → 持久化存儲(chǔ) → 推送給Consumer → 等待ACK → 刪除/重投

// 消費(fèi)端保證示例
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage order, Channel channel, 
                       @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        // 1. 業(yè)務(wù)處理
        orderService.process(order);
        // 2. 手動(dòng)確認(rèn)（成功才ACK）
        channel.basicAck(tag, false);
        // 3. 更新消費(fèi)記錄
        consumeRecordService.markConsumed(order.getId());
    } catch (Exception e) {
        // 4. 失敗處理：重試或進(jìn)入死信隊(duì)列
        if (retryCount < MAX_RETRY) {
            channel.basicNack(tag, false, true); // 重入隊(duì)列
        } else {
            channel.basicNack(tag, false, false); // 進(jìn)入死信隊(duì)列
            alarmService.notifyAdmin(order, e);
        }
    }
}

public boolean processWithIdempotent(String msgId) {
    // 基于消息ID去重
    if (redis.exists("processed:" + msgId)) {
        return true; // 已處理過(guò)
    }
    // 業(yè)務(wù)處理
    boolean success = doBusinessLogic();
    // 記錄處理狀態(tài)
    if (success) {
        redis.setex("processed:" + msgId, 24h, "1");
    }
    return success;
}

兩階段提交：
1. 發(fā)送Half Message（預(yù)備消息）
2. 執(zhí)行本地事務(wù)
3. 根據(jù)本地事務(wù)結(jié)果Commit/Rollback
4. Broker檢查事務(wù)狀態(tài)并投遞/丟棄

# 補(bǔ)償機(jī)制實(shí)現(xiàn)
def reliable_delivery(message):
    max_retries = 5
    for attempt in range(max_retries):
        try:
            # 嘗試投遞
            result = mq_client.send(message)
            if result.confirmed:
                log_delivery_success(message.id)
                return True
        except Exception as e:
            log_failure(attempt, e)
            if attempt == max_retries - 1:
                # 最終失敗，人工介入
                send_alert_to_admin(message)
                save_to_compensation_table(message)
                return False
            # 等待后重試
            sleep(backoff_time(attempt))
    return False

-- 本地消息表結(jié)構(gòu)
CREATE TABLE local_message (
    id BIGINT PRIMARY KEY,
    biz_id VARCHAR(64),      -- 業(yè)務(wù)ID
    content TEXT,           -- 消息內(nèi)容
    status TINYINT,         -- 0:待發(fā)送, 1:已發(fā)送, 2:已確認(rèn)
    retry_count INT,
    next_retry_time DATETIME,
    created_at TIMESTAMP
);

// 批量消息的可靠性處理
public class BatchMessageReliableSender {
    public void sendBatch(List<Message> batch) {
        // 1. 批量持久化到本地
        batchMessageDao.saveAll(batch);
        // 2. 設(shè)置批次ID
        String batchId = generateBatchId();
        // 3. 發(fā)送批次消息
        boolean success = mqTemplate.sendBatch(batchId, batch);
        // 4. 批次確認(rèn)（或單條補(bǔ)償）
        if (success) {
            markBatchDelivered(batchId);
        } else {
            // 逐條重試或記錄異常
            compensateFailedMessages(batch);
        }
    }
}

-- 消息對(duì)賬SQL示例
SELECT 
  DATE(create_time) as day,
  COUNT(*) as total_sent,
  SUM(CASE WHEN status=2 THEN 1 ELSE 0 END) as confirmed,
  SUM(CASE WHEN status=1 THEN 1 ELSE 0 END) as pending
FROM message_record
GROUP BY DATE(create_time)
HAVING total_sent != confirmed;

# 配置示例：多級(jí)降級(jí)
mq:
  primary:
    url: "amqp://primary"
    timeout: 1000ms
  secondary:
    url: "amqp://secondary"
    timeout: 2000ms
  fallback-to-db: true  # 最終降級(jí)到數(shù)據(jù)庫(kù)