分布式系统核心问题

单体架构的限制:
- 代码库庞大,编译部署缓慢
- 单机性能瓶颈,无法满足高并发需求
- 团队协作困难,冲突频繁
- 技术栈受限,无法灵活选择

分布式架构的优势:
- 服务拆分,独立部署和扩展
- 水平扩展,应对流量增长
- 团队独立开发,提高效率
- 技术选型灵活,因地制宜

单体架构:单点故障
┌─────┐
│ App │ ──── 挂掉 → 全系统不可用
└─────┘

分布式架构:容错设计
┌─────┐
│ App │ ──── 挂掉 → 其他节点接管
└─────┘
     ↓
┌─────┐
│ App │ ──── 继续服务
└─────┘

技术栈演进:
单体应用 → 垂直拆分 → 分布式服务 → 微服务架构 → 云原生架构

单机场景:
┌────────────┐
│   应用     │
│  ┌──────┐  │
│  │方法A │  │ ← 调用可靠,立即返回
│  └──────┘  │
│  ┌──────┐  │
│  │方法B │  │ ← 内存访问,稳定
│  └──────┘  │
└────────────┘

分布式场景:
┌─────┐      网络      ┌─────┐
│服务A│ ←──────────────→ │服务B│
└─────┘   ↑ 网络不可靠   └─────┘
          │
          ├─ 超时
          ├─ 丢包
          ├─ 延迟
          ├─ 网络分区
          └─ 部分失败

// 调用远程服务
public User getUser(Long id) {
    try {
        // 超时时间 3 秒
        return userService.getUser(id);
    } catch (TimeoutException e) {
        // 三种情况:
        // 1. 请求未到达服务提供方
        // 2. 服务提供方已执行,但响应超时
        // 3. 服务提供方执行失败,未返回响应
        
        // 如何处理?
        // - 重试?可能导致重复执行
        // - 查询?可能查不到结果
        // - 放弃?可能导致数据不一致
    }
}

// 设置合理的超时时间
public class TimeoutConfig {
    // 连接超时
    private int connectTimeout = 3000; // 3 秒
    
    // 读取超时
    private int readTimeout = 5000; // 5 秒
    
    // 总超时
    private int totalTimeout = 10000; // 10 秒
}

// 超时重试策略
public class RetryPolicy {
    private int maxRetries = 3;        // 最大重试次数
    private long retryInterval = 1000; // 重试间隔
    
    public <T> T execute(Callable<T> task) {
        for (int i = 0; i < maxRetries; i++) {
            try {
                return task.call();
            } catch (TimeoutException e) {
                if (i == maxRetries - 1) {
                    throw new RuntimeException("重试失败", e);
                }
                Thread.sleep(retryInterval);
            }
        }
    }
}

正常情况:
┌──────┐ ←─────→ ┌──────┐
│节点A │         │节点B │
└──────┘ ←─────→ └──────┘
   ↑                 ↑
   └────────┬────────┘
         数据同步

网络分区:
┌──────┐    ╳    ┌──────┐
│节点A │         │节点B │
└──────┘         └──────┘
   ↑                 ↑
   │                 │
   └───── 数据不一致 ─┘

// 1. 多数派原则(Quorum)
// 写入需要多数节点确认
public boolean write(String key, String value) {
    int successCount = 0;
    int totalNodes = 5;
    int quorum = totalNodes / 2 + 1; // 3
    
    for (Node node : nodes) {
        try {
            node.write(key, value);
            successCount++;
            if (successCount >= quorum) {
                return true; // 写入成功
            }
        } catch (Exception e) {
            // 忽略单个节点失败
        }
    }
    return false; // 写入失败
}

// 2. 故障检测和隔离
public class FailureDetector {
    private Map<Node, Long> lastHeartbeat = new ConcurrentHashMap<>();
    
    public void checkNodes() {
        for (Node node : nodes) {
            long lastTime = lastHeartbeat.get(node);
            if (System.currentTimeMillis() - lastTime > timeout) {
                // 标记节点为故障
                markAsFailed(node);
            }
        }
    }
}

// × 错误:没有设置超时
public User getUser(Long id) {
    return userService.getUser(id); // 可能永远阻塞
}

// √ 正确:设置超时
public User getUser(Long id) {
    try {
        return userService.getUser(id)
            .get(3, TimeUnit.SECONDS); // 超时 3 秒
    } catch (TimeoutException e) {
        log.error("调用超时", e);
        return null;
    }
}

// × 错误:失败后直接放弃
public User getUser(Long id) {
    try {
        return userService.getUser(id);
    } catch (Exception e) {
        return null; // 可能只是临时故障
    }
}

// √ 正确:合理重试
@Retryable(
    value = {TimeoutException.class},
    maxAttempts = 3,
    backoff = @Backoff(delay = 1000)
)
public User getUser(Long id) {
    return userService.getUser(id);
}

// × 错误:依赖服务不可用导致整个系统不可用
public User getUser(Long id) {
    return userService.getUser(id); // userService 挂了,整个系统不可用
}

// √ 正确:降级处理
public User getUser(Long id) {
    try {
        return userService.getUser(id);
    } catch (Exception e) {
        // 降级:返回缓存数据或默认值
        return cacheService.getUser(id);
        // 或返回默认用户
        // return User.defaultUser();
    }
}

// 单机事务:简单、可靠
@Transactional
public void transfer(Long fromId, Long toId, BigDecimal amount) {
    // 1. 扣减账户A余额
    accountMapper.decreaseBalance(fromId, amount);
    
    // 2. 增加账户B余额
    accountMapper.increaseBalance(toId, amount);
    
    // 要么全部成功,要么全部回滚
}

// 分布式事务:复杂、不可靠
public void transfer(Long fromId, Long toId, BigDecimal amount) {
    // 1. 扣减账户A余额(服务A)
    accountServiceA.decreaseBalance(fromId, amount);
    
    // 2. 增加账户B余额(服务B)
    accountServiceB.increaseBalance(toId, amount);
    
    // 问题:
    // - 服务A成功,服务B失败 → 数据不一致
    // - 服务A成功,网络超时 → 不知道服务B是否成功
    // - 服务A成功,服务B超时 → 不知道是否成功
}

// 两阶段提交(2PC)
public class TwoPhaseCommit {
    
    // 阶段1:准备阶段
    public boolean prepare(List<Service> services) {
        for (Service service : services) {
            if (!service.prepare()) {
                return false; // 准备失败
            }
        }
        return true;
    }
    
    // 阶段2:提交阶段
    public void commit(List<Service> services) {
        for (Service service : services) {
            service.commit(); // 提交事务
        }
    }
    
    // 阶段2:回滚阶段
    public void rollback(List<Service> services) {
        for (Service service : services) {
            service.rollback(); // 回滚事务
        }
    }
}

// 基于消息的最终一致性
public class OrderService {
    
    @Transactional
    public void createOrder(Order order) {
        // 1. 创建订单
        orderMapper.insert(order);
        
        // 2. 发送消息到消息队列
        mqService.send("order.created", order);
        
        // 订单服务和库存服务异步处理
    }
}

public class InventoryService {
    
    @RabbitListener(queues = "order.created")
    public void handleOrderCreated(Order order) {
        // 异步扣减库存
        inventoryMapper.decreaseStock(order.getProductId(), order.getQuantity());
    }
}

         C (Consistency)
         一致性
        / \
       /   \
      /     \
     /  CAP  \
    /         \
   A ───────── P
可用性      分区容错性

场景:网络分区发生

节点A ──────╳────── 节点B
  │                  │
  │                  │
  ↓                  ↓
客户端1            客户端2

问题:
- 如果选择 C:拒绝客户端请求,保证一致性,但牺牲可用性
- 如果选择 A:接受客户端请求,保证可用性,但牺牲一致性

// 基本可用:降级处理
public User getUser(Long id) {
    try {
        return userService.getUser(id);
    } catch (Exception e) {
        // 基本可用:返回缓存数据
        return cacheService.getUser(id);
    }
}

// 软状态:允许中间状态
public void updateOrderStatus(Long orderId, String status) {
    // 中间状态:订单状态可能短暂不一致
    orderMapper.updateStatus(orderId, status);
    
    // 异步通知其他服务
    mqService.send("order.status.updated", orderId, status);
}

// 最终一致:补偿机制
@Scheduled(fixedDelay = 60000)
public void checkConsistency() {
    // 定期检查数据一致性
    List<Order> inconsistentOrders = orderMapper.findInconsistentOrders();
    
    for (Order order : inconsistentOrders) {
        // 补偿操作
        compensate(order);
    }
}

是否需要强一致性?
├─ 是 → 选择 CP 方案
│       ├─ 数据库主从复制
│       ├─ 分布式锁
│       └─ 两阶段提交
│
└─ 否 → 选择 AP 方案
        ├─ 允许不一致时间窗口?
        │   ├─ 短(秒级) → 最终一致性
        │   │           ├─ 消息队列
        │   │           ├─ 异步复制
        │   │           └─ 补偿机制
        │   │
        │   └─ 长(分钟级) → 弱一致性
        │                   ├─ 定时同步
        │                   └─ 定期补偿

f(x) = f(f(x))

// 消息队列可能重复投递
@RabbitListener(queues = "order.created")
public void handleOrderCreated(Order order) {
    // 问题:如果这条消息重复投递,会导致重复扣减库存
    inventoryService.decreaseStock(order.getProductId(), order.getQuantity());
}

// 客户端超时重试
public void createOrder(Order order) {
    try {
        orderService.createOrder(order);
    } catch (TimeoutException e) {
        // 客户端重试,可能导致重复创建订单
        createOrder(order);
    }
}

// 用户快速点击两次按钮
// 可能导致重复提交
$("#submit").click(function() {
    $.post("/order/create", orderData);
});

// 使用业务单号实现幂等
public class OrderService {
    
    @Transactional
    public void createOrder(Order order) {
        // 1. 检查订单号是否已存在
        Order existingOrder = orderMapper.selectByOrderNo(order.getOrderNo());
        if (existingOrder != null) {
            // 订单已存在,直接返回(幂等)
            return existingOrder;
        }
        
        // 2. 创建订单
        orderMapper.insert(order);
        
        // 3. 扣减库存
        inventoryService.decreaseStock(order.getProductId(), order.getQuantity());
        
        return order;
    }
}

// 客户端生成唯一订单号
public class OrderClient {
    public void createOrder() {
        // 生成唯一订单号
        String orderNo = UUID.randomUUID().toString();
        
        Order order = new Order();
        order.setOrderNo(orderNo);
        order.setProductId(123L);
        order.setQuantity(1);
        
        orderService.createOrder(order);
    }
}

-- 数据库唯一索引
CREATE TABLE orders (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_no VARCHAR(64) UNIQUE,  -- 唯一索引
    product_id BIGINT,
    quantity INT,
    create_time DATETIME
);

-- 幂等插入
INSERT INTO orders (order_no, product_id, quantity, create_time)
VALUES ('ORD123456', 123, 1, NOW())
ON DUPLICATE KEY UPDATE order_no = order_no; -- 重复插入时忽略

// 幂等 Token 机制
public class IdempotentService {
    
    // 1. 获取 Token
    public String getToken() {
        String token = UUID.randomUUID().toString();
        redisTemplate.opsForValue().set(token, "1", 10, TimeUnit.MINUTES);
        return token;
    }
    
    // 2. 验证 Token
    public boolean checkToken(String token) {
        // 使用 Lua 脚本保证原子性
        String script = 
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "    return redis.call('del', KEYS[1]) " +
            "else " +
            "    return 0 " +
            "end";
        
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class),
            Collections.singletonList(token),
            "1"
        );
        
        return result != null && result == 1;
    }
    
    // 3. 业务处理
    public void process(String token, Order order) {
        // 验证 Token
        if (!checkToken(token)) {
            throw new RuntimeException("重复请求");
        }
        
        // 执行业务逻辑
        createOrder(order);
    }
}

// 使用状态机实现幂等
public class OrderStateMachine {
    
    // 订单状态
    public enum OrderStatus {
        CREATED,    // 已创建
        PAID,       // 已支付
        SHIPPED,    // 已发货
        COMPLETED,  // 已完成
        CANCELLED   // 已取消
    }
    
    // 状态转换规则
    private static final Map<OrderStatus, Set<OrderStatus>> TRANSITIONS = new HashMap<>();
    
    static {
        TRANSITIONS.put(CREATED, Set.of(PAID, CANCELLED));
        TRANSITIONS.put(PAID, Set.of(SHIPPED, CANCELLED));
        TRANSITIONS.put(SHIPPED, Set.of(COMPLETED));
    }
    
    // 状态转换
    @Transactional
    public void updateStatus(Long orderId, OrderStatus newStatus) {
        Order order = orderMapper.selectById(orderId);
        
        // 检查状态转换是否合法
        if (!TRANSITIONS.get(order.getStatus()).contains(newStatus)) {
            throw new IllegalStateException("非法状态转换");
        }
        
        // 更新状态
        orderMapper.updateStatus(orderId, newStatus);
    }
    
    // 幂等支付
    public void pay(Long orderId) {
        Order order = orderMapper.selectById(orderId);
        
        // 如果已经是已支付状态,直接返回(幂等)
        if (order.getStatus() == OrderStatus.PAID) {
            return;
        }
        
        // 状态转换
        updateStatus(orderId, OrderStatus.PAID);
    }
}

// 查询操作:天然幂等
public User getUser(Long id) {
    return userMapper.selectById(id);
}

// 删除操作:幂等
public void deleteUser(Long id) {
    userMapper.deleteById(id); // 多次删除结果相同
}

// 更新为固定值:幂等
public void updateStatus(Long id, String status) {
    userMapper.updateStatus(id, status); // 多次更新结果相同
}

// × 非幂等:累加操作
public void increaseBalance(Long id, BigDecimal amount) {
    userMapper.increaseBalance(id, amount); // 多次累加导致余额错误
}

// √ 幂等:使用业务单号
public void increaseBalance(Long id, BigDecimal amount, String transactionNo) {
    // 检查交易号是否已存在
    if (transactionMapper.exists(transactionNo)) {
        return; // 已处理,幂等
    }
    
    // 增加余额
    userMapper.increaseBalance(id, amount);
    
    // 记录交易
    transactionMapper.insert(transactionNo, amount);
}

-- 基于数据库的唯一索引
CREATE TABLE distributed_lock (
    lock_key VARCHAR(64) PRIMARY KEY,
    lock_value VARCHAR(64),
    expire_time DATETIME
);

-- 加锁
INSERT INTO distributed_lock (lock_key, lock_value, expire_time)
VALUES ('lock:order:123', 'uuid', DATE_ADD(NOW(), INTERVAL 30 SECOND));

-- 解锁
DELETE FROM distributed_lock 
WHERE lock_key = 'lock:order:123' AND lock_value = 'uuid';

// Redis 分布式锁
public class RedisDistributedLock {
    
    private StringRedisTemplate redisTemplate;
    
    // 加锁
    public boolean lock(String key, String value, long expireTime) {
        // 使用 SETNX 命令
        Boolean result = redisTemplate.opsForValue()
            .setIfAbsent(key, value, expireTime, TimeUnit.SECONDS);
        
        return result != null && result;
    }
    
    // 解锁
    public boolean unlock(String key, String value) {
        // 使用 Lua 脚本保证原子性
        String script = 
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "    return redis.call('del', KEYS[1]) " +
            "else " +
            "    return 0 " +
            "end";
        
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class),
            Collections.singletonList(key),
            value
        );
        
        return result != null && result == 1;
    }
    
    // 使用示例
    public void processOrder(Long orderId) {
        String lockKey = "lock:order:" + orderId;
        String lockValue = UUID.randomUUID().toString();
        
        try {
            // 加锁
            if (lock(lockKey, lockValue, 30)) {
                // 执行业务逻辑
                createOrder(orderId);
            } else {
                throw new RuntimeException("获取锁失败");
            }
        } finally {
            // 解锁
            unlock(lockKey, lockValue);
        }
    }
}

// Zookeeper 分布式锁
public class ZkDistributedLock {
    
    private CuratorFramework client;
    
    // 加锁
    public InterProcessMutex acquire(String lockPath, long timeout, TimeUnit unit) 
            throws Exception {
        InterProcessMutex lock = new InterProcessMutex(client, lockPath);
        
        if (lock.acquire(timeout, unit)) {
            return lock;
        } else {
            throw new RuntimeException("获取锁超时");
        }
    }
    
    // 使用示例
    public void processOrder(Long orderId) throws Exception {
        String lockPath = "/locks/order/" + orderId;
        InterProcessMutex lock = null;
        
        try {
            // 加锁
            lock = acquire(lockPath, 10, TimeUnit.SECONDS);
            
            // 执行业务逻辑
            createOrder(orderId);
        } finally {
            // 解锁
            if (lock != null) {
                lock.release();
            }
        }
    }
}

// × 错误:使用分布式锁
public void createOrder(Order order) {
    String lockKey = "lock:order:" + order.getUserId();
    distributedLock.lock(lockKey);
    
    try {
        // 创建订单
        orderMapper.insert(order);
    } finally {
        distributedLock.unlock(lockKey);
    }
}

// √ 正确:使用幂等设计
public void createOrder(Order order) {
    // 检查订单号是否已存在
    if (orderMapper.exists(order.getOrderNo())) {
        return; // 幂等
    }
    
    // 创建订单
    orderMapper.insert(order);
}

//  分布式锁的问题:
// 1. Redis 单节点故障,锁丢失
// 2. 客户端获取锁后崩溃,锁无法释放
// 3. 锁超时时间设置不合理

// 解决方案:
// 1. 使用 RedLock 算法(多节点)
// 2. 设置合理的超时时间
// 3. 使用看门狗机制自动续期

// × 错误:长事务 + 分布式锁
public void processOrder(Long orderId) {
    distributedLock.lock("order:" + orderId);
    
    try {
        // 长时间业务处理
        createOrder(orderId);      // 1 秒
        decreaseInventory(orderId); // 2 秒
        sendEmail(orderId);         // 3 秒
        // 总共 6 秒,可能超过锁超时时间
    } finally {
        distributedLock.unlock("order:" + orderId);
    }
}

// √ 正确:缩短锁持有时间
public void processOrder(Long orderId) {
    // 只锁关键操作
    distributedLock.lock("inventory:" + productId);
    try {
        decreaseInventory(orderId);
    } finally {
        distributedLock.unlock("inventory:" + productId);
    }
    
    // 其他操作不需要锁
    createOrder(orderId);
    sendEmail(orderId);
}

// 设置合理的超时时间
public class LockConfig {
    // 业务执行时间 + 缓冲时间
    private long lockTimeout = businessTimeout * 1.5;
}

// × 粗粒度锁
String lockKey = "lock:order";

// √ 细粒度锁
String lockKey = "lock:order:" + orderId;

// 带重试的加锁
public boolean lockWithRetry(String key, String value, long expireTime, int maxRetries) {
    for (int i = 0; i < maxRetries; i++) {
        if (lock(key, value, expireTime)) {
            return true;
        }
        
        try {
            Thread.sleep(100); // 等待后重试
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
    return false;
}

// Redisson 看门狗机制
RLock lock = redisson.getLock("myLock");
lock.lock(); // 默认 30 秒,看门狗自动续期

// 业务执行期间,看门狗会自动续期
// 业务执行完毕,主动释放锁
lock.unlock();

// × 错误:无限重试
public void callService() {
    while (true) {
        try {
            remoteService.call();
            break;
        } catch (Exception e) {
            // 无限重试,可能导致系统崩溃
            continue;
        }
    }
}

// √ 正确:有限重试 + 幂等设计
@Retryable(
    value = {TimeoutException.class},
    maxAttempts = 3,
    backoff = @Backoff(delay = 1000, multiplier = 2)
)
public void callService(String requestId) {
    // 使用 requestId 实现幂等
    remoteService.call(requestId);
}

电商系统的不同场景:

1. 下单场景
   - 需求:强一致性(不能超卖)
   - 选择:CP 方案
   - 实现:分布式锁 + 数据库事务

2. 商品浏览场景
   - 需求:高可用性
   - 选择:AP 方案
   - 实现:缓存 + 异步更新

3. 订单查询场景
   - 需求:最终一致性
   - 选择:AP 方案
   - 实现:读写分离 + 异步复制

// × 错误:只发消息,没有补偿
public void createOrder(Order order) {
    // 创建订单
    orderMapper.insert(order);
    
    // 发送消息
    mqService.send("order.created", order);
    
    // 如果消息丢失,库存永远不会扣减
}

// √ 正确:增加补偿机制
public void createOrder(Order order) {
    // 创建订单
    orderMapper.insert(order);
    
    // 发送消息
    mqService.send("order.created", order);
}

// 定时补偿任务
@Scheduled(fixedDelay = 60000)
public void compensateOrders() {
    // 查找未处理的订单
    List<Order> pendingOrders = orderMapper.findPendingOrders();
    
    for (Order order : pendingOrders) {
        // 重新发送消息
        mqService.send("order.created", order);
    }
}

Paxos 算法 vs Raft 算法:

Paxos 算法:
- 理论:正确性已证明
- 实际:难以理解和实现
- 落地:很少有团队能正确实现

Raft 算法:
- 理论:正确性已证明
- 实际:易于理解和实现
- 落地:被广泛使用(Etcd、Consul)

结论:选择适合团队能力的方案

// TCC (Try-Confirm-Cancel) 分布式事务
public class OrderTccService {
    
    // Try 阶段:预留资源
    public void tryCreateOrder(Order order) {
        // 1. 创建订单(待确认状态)
        orderMapper.insert(order.setStatus("TRYING"));
        
        // 2. 预留库存
        inventoryService.freezeStock(order.getProductId(), order.getQuantity());
        
        // 3. 预留余额
        accountService.freezeBalance(order.getUserId(), order.getAmount());
    }
    
    // Confirm 阶段:确认提交
    public void confirmCreateOrder(Order order) {
        // 1. 确认订单
        orderMapper.updateStatus(order.getId(), "CONFIRMED");
        
        // 2. 扣减库存
        inventoryService.decreaseStock(order.getProductId(), order.getQuantity());
        
        // 3. 扣减余额
        accountService.decreaseBalance(order.getUserId(), order.getAmount());
    }
    
    // Cancel 阶段:取消回滚
    public void cancelCreateOrder(Order order) {
        // 1. 取消订单
        orderMapper.updateStatus(order.getId(), "CANCELLED");
        
        // 2. 释放库存
        inventoryService.unfreezeStock(order.getProductId(), order.getQuantity());
        
        // 3. 释放余额
        accountService.unfreezeBalance(order.getUserId(), order.getAmount());
    }
}

// 基于消息的最终一致性
public class OrderService {
    
    @Transactional
    public void createOrder(Order order) {
        // 1. 创建订单
        orderMapper.insert(order);
        
        // 2. 发送消息到库存服务
        InventoryEvent event = new InventoryEvent();
        event.setOrderId(order.getId());
        event.setProductId(order.getProductId());
        event.setQuantity(order.getQuantity());
        
        // 保存消息到本地消息表
        eventMapper.insert(event);
        
        // 异步发送消息
        mqService.send("inventory.decrease", event);
    }
}

// 库存服务消费消息
public class InventoryService {
    
    @RabbitListener(queues = "inventory.decrease")
    public void handleDecreaseStock(InventoryEvent event) {
        // 幂等检查
        if (eventMapper.exists(event.getId())) {
            return; // 已处理
        }
        
        // 扣减库存
        inventoryMapper.decreaseStock(event.getProductId(), event.getQuantity());
        
        // 标记消息已处理
        eventMapper.markAsProcessed(event.getId());
    }
}

// 使用分布式锁实现任务调度
public class ScheduledTaskService {
    
    @Scheduled(cron = "0 0 2 * * ?")
    public void executeTask() {
        String lockKey = "lock:task:daily-cleanup";
        String lockValue = UUID.randomUUID().toString();
        
        // 获取分布式锁
        if (distributedLock.lock(lockKey, lockValue, 300)) {
            try {
                // 执行任务
                doCleanup();
            } finally {
                // 释放锁
                distributedLock.unlock(lockKey, lockValue);
            }
        } else {
            log.info("其他实例正在执行任务");
        }
    }
    
    private void doCleanup() {
        // 清理过期数据
        // 生成报表
        // 发送通知
    }
}

// 消息幂等处理
@RabbitListener(queues = "order.created")
public void handleOrderCreated(OrderEvent event) {
    // 幂等检查
    if (eventMapper.exists(event.getEventId())) {
        log.info("消息已处理: {}", event.getEventId());
        return;
    }
    
    // 处理订单
    createOrder(event);
    
    // 标记消息已处理
    eventMapper.insert(event.getEventId());
}

// 顺序消息:同一订单的消息发送到同一队列
public void sendOrderMessage(Long orderId, OrderEvent event) {
    // 根据 orderId 计算队列编号
    int queueIndex = (int) (orderId % queueCount);
    
    String queueName = "order.queue." + queueIndex;
    
    rabbitTemplate.convertAndSend(queueName, event);
}

// 消息确认机制
@RabbitListener(queues = "order.created")
public void handleOrderCreated(OrderEvent event, Channel channel, 
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        // 处理订单
        createOrder(event);
        
        // 确认消息
        channel.basicAck(tag, false);
    } catch (Exception e) {
        // 拒绝消息,重新入队
        channel.basicNack(tag, false, true);
    }
}

// 防重:Token 机制
if (!checkToken(token)) {
    throw new RuntimeException("重复请求");
}

// 幂等:业务单号
if (orderMapper.exists(orderNo)) {
    return; // 已存在,直接返回
}
createOrder(order);

// 设置合理的超时时间
public class TimeoutConfig {
    private int connectTimeout = 3000;  // 连接超时
    private int readTimeout = 5000;     // 读取超时
    private int totalTimeout = 10000;   // 总超时
}

// 优先选择业务约束
if (order.getStatus() == PAID) {
    return; // 已支付,幂等
}

// 其次选择状态机
if (!canTransition(order.getStatus(), newStatus)) {
    throw new IllegalStateException("非法状态转换");
}

// 最后选择分布式锁
distributedLock.lock("order:" + orderId);