当前位置：首页 > article >正文

【RabbitMQ业务幂等设计】RabbitMQ消息是幂等的吗？

article 2025/2/23 4:24:50

在分布式系统中，RabbitMQ 自身不直接提供消息幂等性保障机制，但可通过业务逻辑设计和技术组合实现消息处理的幂等性。以下是 8 种核心实现方案及最佳实践：

一、消息唯一标识符 (Message Deduplication)

原理
- 每条消息携带全局唯一ID（如 UUID、Snowflake ID）
- 消费者维护已处理消息ID的存储（Redis/DB）

实现步骤

// 生产者端
MessageProperties props = new MessageProperties();
props.setMessageId(UUID.randomUUID().toString());
Message message = new Message(body.getBytes(), props);

// 消费者端
@RabbitListener(queues = "order_queue")
public void process(Message message) {
    String msgId = message.getMessageProperties().getMessageId();
    if (redis.setnx(msgId, "processed") == 1) {
        // 处理业务逻辑
        // 成功后设置过期时间防止存储膨胀
        redis.expire(msgId, 72 * 3600); 
    } else {
        // 幂等拦截
    }
}

二、版本号控制 (Optimistic Concurrency Control)

适用场景
数据更新类操作（如账户余额修改）

实现方案

-- 消息体包含数据版本号
UPDATE account 
SET balance = new_balance, version = version + 1 
WHERE id = 123 AND version = current_version;

三、状态机驱动 (State Machine)

应用场景
订单状态流转（创建→支付→发货）

实现示例

public void handleOrderMessage(OrderMessage msg) {
    Order order = orderDao.get(msg.getOrderId());
    if (order.getStatus() != msg.getExpectedStatus()) {
        log.warn("状态不匹配，当前状态:{}", order.getStatus());
        return;
    }
    // 执行状态变更逻辑
}

四、业务唯一键约束

实现方式

CREATE TABLE payment_records (
  id BIGINT PRIMARY KEY,
  order_no VARCHAR(64) UNIQUE, -- 业务唯一键
  amount DECIMAL(10,2)
);

-- 插入时捕获唯一键冲突
try {
    insertPaymentRecord();
} catch (DuplicateKeyException e) {
    // 幂等处理
}

五、消息确认策略优化

关键配置

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual  # 手动ACK
        retry:
          enabled: true
          max-attempts: 3         # 最大重试次数

处理逻辑

@RabbitListener(queues = "critical_queue")
public void process(Message message, Channel channel) throws IOException {
    try {
        // 业务处理
        channel.basicAck(tag, false);
    } catch (Exception e) {
        channel.basicNack(tag, false, false); // 直接进入死信队列
    }
}

六、分布式锁机制

Redis 分布式锁示例

public void processWithLock(Message msg) {
    String lockKey = "msg_lock:" + msg.getId();
    try {
        if (redisLock.tryLock(lockKey, 30)) {
            // 真正的业务处理
        }
    } finally {
        redisLock.unlock(lockKey);
    }
}

七、时序控制 (Timestamp Validation)

实现逻辑

if (message.getEventTime() < lastProcessedTime.get()) {
    log.info("丢弃过期消息，事件时间:{}", message.getEventTime());
    return;
}

八、消息轨迹追踪表

设计表结构

CREATE TABLE message_log (
  message_id VARCHAR(64) PRIMARY KEY,
  status ENUM('PROCESSING','SUCCESS','FAILED'),
  processed_time DATETIME,
  retry_count INT DEFAULT 0
);

处理流程

// 开启事务
beginTransaction();
try {
    // 1. 插入消息记录
    insertMessageLog(msgId, "PROCESSING");
    
    // 2. 执行业务操作
    processBusinessLogic();
    
    // 3. 更新状态
    updateMessageStatus(msgId, "SUCCESS");
    commit();
} catch (Exception e) {
    rollback();
}