消息队列如何保证可靠性

// 库存服务处理消息的代码示例
public void processInventoryReduction(InventoryMessage message) {
    String messageId = message.getMessageId(); // 消息唯一ID
    
    // 检查消息是否已处理
    if (messageProcessRepository.isProcessed(messageId)) {
        log.info("消息{}已处理，忽略重复消息", messageId);
        return;
    }
    
    try {
        // 执行库存扣减逻辑
        inventoryService.reduce(message.getProductId(), message.getQuantity());
        
        // 标记消息为已处理
        messageProcessRepository.markAsProcessed(messageId);
    } catch (Exception e) {
        log.error("处理消息{}失败", messageId, e);
        throw e;
    }
}

-- 消息处理记录表
CREATE TABLE message_process_record (
    message_id VARCHAR(50) PRIMARY KEY,  -- 消息唯一ID作为主键确保唯一性
    consumer_group VARCHAR(50) NOT NULL, -- 消费者组标识
    process_time TIMESTAMP NOT NULL,     -- 处理时间
    process_status VARCHAR(20) NOT NULL  -- 处理状态
);

-- 可选：添加索引优化查询
CREATE INDEX idx_consumer_message ON message_process_record(consumer_group, message_id);

@Service
public class InventoryConsumer {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    @Autowired
    private InventoryService inventoryService;
    
    // 消费者组标识，通常从配置中获取
    private static final String CONSUMER_GROUP = "inventory-service";
    
    @RabbitListener(queues = "inventory-queue")
    public void processInventoryReduction(InventoryMessage message) {
        String messageId = message.getMessageId();
        
        // 检查消息是否已处理 - 使用共享数据库
        if (isMessageProcessed(messageId)) {
            log.info("消息{}已处理，忽略重复消息", messageId);
            return;
        }
        
        try {
            // 尝试记录消息处理状态 - 使用数据库唯一约束确保幂等
            if (!markMessageProcessing(messageId)) {
                log.info("消息{}已被其他实例处理，忽略", messageId);
                return;
            }
            
            // 执行库存扣减逻辑
            inventoryService.reduce(message.getProductId(), message.getQuantity());
            
            // 标记消息为已成功处理
            markMessageProcessed(messageId);
        } catch (Exception e) {
            log.error("处理消息{}失败", messageId, e);
            // 可选：标记消息处理失败，以便后续分析
            markMessageFailed(messageId, e.getMessage());
            throw e;
        }
    }
    
    private boolean isMessageProcessed(String messageId) {
        int count = jdbcTemplate.queryForObject(
            "SELECT COUNT(*) FROM message_process_record WHERE message_id = ? AND consumer_group = ?",
            Integer.class, messageId, CONSUMER_GROUP);
        return count > 0;
    }
    
    private boolean markMessageProcessing(String messageId) {
        try {
            jdbcTemplate.update(
                "INSERT INTO message_process_record (message_id, consumer_group, process_time, process_status) " +
                "VALUES (?, ?, NOW(), 'PROCESSING')",
                messageId, CONSUMER_GROUP);
            return true;
        } catch (DuplicateKeyException e) {
            // 如果消息ID已存在，说明已被处理或正在处理
            return false;
        }
    }
    
    private void markMessageProcessed(String messageId) {
        jdbcTemplate.update(
            "UPDATE message_process_record SET process_status = 'PROCESSED', process_time = NOW() " +
            "WHERE message_id = ? AND consumer_group = ?",
            messageId, CONSUMER_GROUP);
    }
    
    private void markMessageFailed(String messageId, String errorMessage) {
        jdbcTemplate.update(
            "UPDATE message_process_record SET process_status = 'FAILED', error_message = ?, process_time = NOW() " +
            "WHERE message_id = ? AND consumer_group = ?",
            errorMessage, messageId, CONSUMER_GROUP);
    }
}

@Autowired
private RedissonClient redissonClient;

public void processInventoryReduction(InventoryMessage message) {
    String messageId = message.getMessageId();
    String lockKey = "inventory_msg_lock:" + messageId;
    
    // 检查消息是否已处理
    if (isMessageProcessed(messageId)) {
        log.info("消息{}已处理，忽略重复消息", messageId);
        return;
    }
    
    // 获取分布式锁，确保同一时间只有一个实例处理该消息
    RLock lock = redissonClient.getLock(lockKey);
    try {
        // 尝试获取锁，等待2秒，持有锁10秒
        if (lock.tryLock(2, 10, TimeUnit.SECONDS)) {
            try {
                // 再次检查，防止锁等待期间其他实例已处理
                if (isMessageProcessed(messageId)) {
                    log.info("获取锁后再次检查：消息{}已处理", messageId);
                    return;
                }
                
                // 标记为处理中
                markMessageProcessing(messageId);
                
                // 执行业务逻辑
                inventoryService.reduce(message.getProductId(), message.getQuantity());
                
                // 标记为已处理
                markMessageProcessed(messageId);
            } finally {
                lock.unlock(); // 确保释放锁
            }
        } else {
            log.info("无法获取消息{}的处理锁，跳过处理", messageId);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        log.error("获取锁被中断", e);
    }
}

def retry_task(order_id):
    redis_client.lpush("email_queue", order_id)  # 重新推入主队列