我们来说一下消息的可靠性投递

// 伪代码示例：生产端确认模式
public void sendWithConfirm(Message msg) {
    // 1. 持久化到本地数据库（防丢失）
    messageDao.save(msg);
    
    // 2. 发送到消息队列
    String msgId = rabbitTemplate.convertAndSend(msg);
    
    // 3. 等待Broker确认
    boolean ack = waitForAck(msgId, TIMEOUT);
    
    // 4. 失败重试（指数退避）
    if (!ack) {
        retryWithBackoff(msg);
    }
    
    // 5. 最终记录投递状态
    updateDeliveryStatus(msgId, ack);
}

消息处理流程：
Producer → Broker接收 → 持久化存储 → 推送给Consumer → 等待ACK → 删除/重投

// 消费端保证示例
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage order, Channel channel, 
                       @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        // 1. 业务处理
        orderService.process(order);
        
        // 2. 手动确认（成功才ACK）
        channel.basicAck(tag, false);
        
        // 3. 更新消费记录
        consumeRecordService.markConsumed(order.getId());
        
    } catch (Exception e) {
        // 4. 失败处理：重试或进入死信队列
        if (retryCount < MAX_RETRY) {
            channel.basicNack(tag, false, true); // 重入队列
        } else {
            channel.basicNack(tag, false, false); // 进入死信队列
            alarmService.notifyAdmin(order, e);
        }
    }
}

public boolean processWithIdempotent(String msgId) {
    // 基于消息ID去重
    if (redis.exists("processed:" + msgId)) {
        return true; // 已处理过
    }
    
    // 业务处理
    boolean success = doBusinessLogic();
    
    // 记录处理状态
    if (success) {
        redis.setex("processed:" + msgId, 24h, "1");
    }
    
    return success;
}

两阶段提交：
1. 发送Half Message（预备消息）
2. 执行本地事务
3. 根据本地事务结果Commit/Rollback
4. Broker检查事务状态并投递/丢弃

# 补偿机制实现
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

-- 本地消息表结构
CREATE TABLE local_message (
    id BIGINT PRIMARY KEY,
    biz_id VARCHAR(64),      -- 业务ID
    content TEXT,           -- 消息内容
    status TINYINT,         -- 0:待发送, 1:已发送, 2:已确认
    retry_count INT,
    next_retry_time DATETIME,
    created_at TIMESTAMP
);

// 批量消息的可靠性处理
public class BatchMessageReliableSender {
    
    public void sendBatch(List<Message> batch) {
        // 1. 批量持久化到本地
        batchMessageDao.saveAll(batch);
        
        // 2. 设置批次ID
        String batchId = generateBatchId();
        
        // 3. 发送批次消息
        boolean success = mqTemplate.sendBatch(batchId, batch);
        
        // 4. 批次确认（或单条补偿）
        if (success) {
            markBatchDelivered(batchId);
        } else {
            // 逐条重试或记录异常
            compensateFailedMessages(batch);
        }
    }
}

-- 消息对账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;

# 配置示例：多级降级
mq:
  primary:
    url: "amqp://primary"
    timeout: 1000ms
  secondary:
    url: "amqp://secondary"
    timeout: 2000ms
  fallback-to-db: true  # 最终降级到数据库