【阶段 11】架构进阶：分布式系统设计与 DDD

<dependencies>
    <!-- 阶段 4 已有的：web / validation / mybatis-plus / mysql / lombok -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <!-- Redisson：生产级分布式锁，内置看门狗续期 -->
    <dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson-spring-boot-starter</artifactId>
        <version>3.27.2</version>
    </dependency>
    <!-- AOP：幂等注解切面需要 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
    <!-- ShardingSphere-JDBC：分库分表（第 4 章演示，可选引入） -->
    <dependency>
        <groupId>org.apache.shardingsphere</groupId>
        <artifactId>shardingsphere-jdbc-core</artifactId>
        <version>5.4.1</version>
    </dependency>
</dependencies>

spring:
  data:
    redis:
      host: localhost
      port: 6379
  datasource:
    url: jdbc:mysql://localhost:3306/dk_mall?useSSL=false&serverTimezone=Asia/Shanghai
    username: root
    password: root

# 启动 Redis（Docker）
docker run -d --name redis -p 6379:6379 redis:7
# 启动 MySQL（Docker，密码 root）
docker run -d --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root mysql:8.0

public enum ErrorCode {
    SUCCESS(0, "success"),
    PARAM_ERROR(40001, "参数错误"),
    UNAUTHORIZED(40101, "未登录或登录已过期"),
    FORBIDDEN(40301, "无权限访问"),
    USER_NOT_FOUND(40401, "用户不存在"),
    DUPLICATE_SUBMIT(40901, "重复提交，请勿重复操作"),
    TOO_MANY_REQUESTS(42901, "请求过于频繁，请稍后重试"),
    SOLD_OUT(41001, "商品已售罄"),
    SYSTEM_ERROR(50000, "系统繁忙，请稍后重试"),
    ;

    private final int code;
    private final String message;

    ErrorCode(int code, String message) {
        this.code = code;
        this.message = message;
    }
    public int getCode() { return code; }
    public String getMessage() { return message; }
}

// 单机互斥：JVM 内的锁，只能锁住同一进程内的线程
public class SingleMachineStock {
    private final ReentrantLock lock = new ReentrantLock();
    private int stock = 100;

    public void deduct() {
        lock.lock();
        try {
            if (stock <= 0) {
                throw new BizException(ErrorCode.SOLD_OUT);
            }
            stock--;
        } finally {
            lock.unlock();
        }
    }
}

SET lock:order:1001 <唯一值> NX EX 30

@Slf4j
@Service
@RequiredArgsConstructor
public class SimpleDistributedLock {

    private final StringRedisTemplate redisTemplate;

    /**
     * 尝试加锁（最小可用版）。
     *
     * @param lockKey   锁的 key，一般是「业务前缀 + 业务 id」，如 lock:order:1001
     * @param requestId 本次请求的唯一标识，解锁时校验「是不是我加的锁」用
     * @param expireSeconds 锁的超时时间，防止持锁进程崩溃后死锁
     * @return true=加锁成功，false=别人正持有锁
     */
    public boolean tryLock(String lockKey, String requestId, long expireSeconds) {
        // setIfAbsent 对应 SET NX EX：NX 保证互斥，EX 保证防死锁，两者原子完成
        Boolean ok = redisTemplate.opsForValue()
                .setIfAbsent(lockKey, requestId, expireSeconds, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(ok);
    }
}

String requestId = UUID.randomUUID().toString();

// 反例：先 GET 判断，再 DEL 删除，两步之间可能被插队，导致误删
public void unlockWrong(String lockKey, String requestId) {
    String value = redisTemplate.opsForValue().get(lockKey);
    if (requestId.equals(value)) {          // 判断：锁是我的
        redisTemplate.delete(lockKey);      // 删除：但这两步之间，锁可能已经过期并被别人拿走
    }
}

-- KEYS[1] = 锁的 key，ARGV[1] = 当前请求的 requestId
-- 只有当锁的 value 等于我的 requestId 时才删除，否则返回 0（不是我加的锁，不动它）
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end

@Slf4j
@Service
@RequiredArgsConstructor
public class RedisDistributedLock {

    private final StringRedisTemplate redisTemplate;

    /** 解锁 Lua 脚本：校验 value + 删除，两步在 Redis 服务端原子完成，防止误删别人的锁 */
    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT = new DefaultRedisScript<>(
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "    return redis.call('del', KEYS[1]) " +
            "else " +
            "    return 0 " +
            "end",
            Long.class
    );

    public boolean tryLock(String lockKey, String requestId, long expireSeconds) {
        Boolean ok = redisTemplate.opsForValue()
                .setIfAbsent(lockKey, requestId, expireSeconds, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(ok);
    }

    /**
     * 解锁：用 Lua 保证「校验 requestId + 删除」原子。
     * 返回 true 表示成功释放；返回 false 表示锁已过期或被别人持有（不处理，避免误删）。
     */
    public boolean unlock(String lockKey, String requestId) {
        Long result = redisTemplate.execute(UNLOCK_SCRIPT,
                Collections.singletonList(lockKey), requestId);
        return result != null && result == 1L;
    }
}

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        // 单机模式；生产建议替换为 useSentinelServers() 或 useClusterServers() 以支持高可用
        config.useSingleServer()
                .setAddress("redis://localhost:6379")
                .setConnectionPoolSize(16);
        return Redisson.create(config);
    }
}

@Slf4j
@Service
@RequiredArgsConstructor
public class RedissonLockService {

    private final RedissonClient redissonClient;

    public void doWithLock(String lockKey, Runnable business) {
        RLock lock = redissonClient.getLock(lockKey);
        boolean locked = false;
        try {
            // tryLock(waitTime, leaseTime, unit)：
            //   waitTime  = 3s：最多等 3 秒拿不到锁就放弃，避免请求无限堆积
            //   leaseTime = -1（不传第三个参数，默认 -1）：启用看门狗，默认 30 秒租期，每 10 秒自动续期
            locked = lock.tryLock(3, TimeUnit.SECONDS);
            if (!locked) {
                // 拿不到锁说明并发竞争激烈，直接快速失败，让上层走限流/降级
                throw new BizException(ErrorCode.TOO_MANY_REQUESTS);
            }
            // 拿到锁，执行业务
            business.run();
        } catch (InterruptedException e) {
            // 等待锁期间被打断，恢复中断标记并抛业务异常
            Thread.currentThread().interrupt();
            throw new BizException(ErrorCode.SYSTEM_ERROR);
        } finally {
            // isHeldByCurrentThread()：只释放「当前线程持有的锁」，防止误删其他线程/其他实例的锁
            if (locked && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

@Slf4j
@Service
@RequiredArgsConstructor
public class StockService {

    private final RedissonClient redissonClient;
    private final StockMapper stockMapper;

    /**
     * 扣减库存（分布式锁保护）。
     * 场景：多实例部署下，防止两个实例同时扣同一商品库存导致超卖。
     */
    public void deductStock(Long productId, int quantity) {
        String lockKey = "lock:stock:" + productId;
        RLock lock = redissonClient.getLock(lockKey);
        boolean locked = false;
        try {
            // 等 2 秒，拿不到锁说明该商品正在被其他请求处理，快速失败
            locked = lock.tryLock(2, TimeUnit.SECONDS);
            if (!locked) {
                log.warn("获取分布式锁失败, productId={}", productId);
                throw new BizException(ErrorCode.TOO_MANY_REQUESTS);
            }
            // 临界区：检查库存 + 扣减。锁内仍需乐观锁兜底（见第 2 章幂等与第 4 章）
            int stock = stockMapper.selectStock(productId);
            if (stock < quantity) {
                throw new BizException(ErrorCode.SOLD_OUT);
            }
            stockMapper.deduct(productId, quantity);
            log.info("扣减库存成功, productId={}, quantity={}", productId, quantity);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BizException(ErrorCode.SYSTEM_ERROR);
        } finally {
            // 只释放当前线程持有的锁，防止误删
            if (locked && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

// 锁内只做临界区，耗时动作移出锁外
public void deductStockOptimized(Long productId, int quantity) {
    RLock lock = redissonClient.getLock("lock:stock:" + productId);
    boolean locked = false;
    try {
        locked = lock.tryLock(2, TimeUnit.SECONDS);
        if (!locked) {
            throw new BizException(ErrorCode.TOO_MANY_REQUESTS);
        }
        // 临界区（短）：一条乐观锁 SQL 搞定，不查库存再扣（省一次往返）
        int rows = stockMapper.deduct(productId, quantity);
        if (rows <= 0) {
            throw new BizException(ErrorCode.SOLD_OUT);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new BizException(ErrorCode.SYSTEM_ERROR);
    } finally {
        if (locked && lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }
    // 锁外（耗时）：发通知、写审计日志、发消息，不占用锁
    notifyService.sendDeductNotice(productId, quantity);
}

// 反例：非幂等。同一请求重试一次，就多插入一条订单
public void createOrder(OrderDTO dto) {
    Order order = new Order();
    order.setOrderNo(dto.getOrderNo());
    order.setAmount(dto.getAmount());
    orderMapper.insert(order);   // 每次调用都插一条，重试就重复
}

CREATE TABLE t_order (
    id          BIGINT PRIMARY KEY,
    order_no    VARCHAR(64) NOT NULL,
    user_id     BIGINT NOT NULL,
    amount      DECIMAL(10,2) NOT NULL,
    status      TINYINT NOT NULL DEFAULT 0,
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_order_no (order_no)   -- 唯一索引：幂等的最后一道防线
) ENGINE=InnoDB;

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderMapper orderMapper;

    public void createOrder(OrderDTO dto) {
        try {
            orderMapper.insert(toEntity(dto));
        } catch (DuplicateKeyException e) {
            // 唯一索引冲突 = 该订单已存在，说明是重复请求，直接吞掉当作成功（幂等）
            log.info("订单已存在，忽略重复提交, orderNo={}", dto.getOrderNo());
        }
    }
}

public enum OrderStatus {
    PENDING(0),   // 待支付
    PAID(1),      // 已支付
    SHIPPED(2),   // 已发货
    CLOSED(3);    // 已关闭

    private final int code;
    OrderStatus(int code) { this.code = code; }

    // 状态机：定义每个状态允许流转到的目标状态
    private static final Map<OrderStatus, Set<OrderStatus>> ALLOWED = Map.of(
            PENDING, Set.of(PAID, CLOSED),
            PAID,    Set.of(SHIPPED),
            SHIPPED, Set.of(),
            CLOSED,  Set.of()
    );

    public boolean canTransitTo(OrderStatus target) {
        return ALLOWED.getOrDefault(this, Set.of()).contains(target);
    }

    public static OrderStatus of(int code) {
        return Arrays.stream(values()).filter(s -> s.code == code)
                .findFirst().orElseThrow(() -> new BizException(ErrorCode.PARAM_ERROR));
    }
}

public void markPaid(Long orderId) {
    Order order = orderMapper.selectById(orderId);
    OrderStatus current = OrderStatus.of(order.getStatus());
    // 状态机校验：已支付的订单不能再次「已支付」，天然幂等
    if (!current.canTransitTo(OrderStatus.PAID)) {
        log.info("订单状态不允许流转，忽略重复操作, orderId={}, current={}", orderId, current);
        return;
    }
    // 用乐观锁更新，where status = 旧状态，防止并发下两个线程同时流转成功
    orderMapper.updateStatus(orderId, current.getCode(), OrderStatus.PAID.getCode());
}

@Update("UPDATE t_order SET status = #{target} WHERE id = #{id} AND status = #{current}")
int updateStatus(@Param("id") Long id, @Param("current") int current, @Param("target") int target);

@Data
public class SubmitOrderDTO {

    /** 请求唯一标识：客户端生成（如 UUID），同一笔业务多次重试必须传同一个值 */
    @NotBlank(message = "requestNo 不能为空")
    private String requestNo;

    @NotNull(message = "商品 id 不能为空")
    private Long productId;

    @NotNull(message = "数量不能为空")
    @Min(value = 1, message = "数量至少为 1")
    private Integer quantity;
}

@Slf4j
@Service
@RequiredArgsConstructor
public class IdempotentSubmitService {

    private final StringRedisTemplate redisTemplate;

    private static final String IDEMPOTENT_KEY_PREFIX = "idempotent:order:";
    /** 幂等记录保留 10 分钟：覆盖「客户端重试」的合理时间窗口，避免 key 永久堆积 */
    private static final long EXPIRE_MINUTES = 10;

    public ApiResult<Void> submit(SubmitOrderDTO dto) {
        String key = IDEMPOTENT_KEY_PREFIX + dto.getRequestNo();
        // setIfAbsent 原子占坑：只有第一个请求能占成功，后续同名 requestNo 全部被拒
        Boolean ok = redisTemplate.opsForValue()
                .setIfAbsent(key, "1", EXPIRE_MINUTES, TimeUnit.MINUTES);
        if (!Boolean.TRUE.equals(ok)) {
            log.info("重复请求被拦截, requestNo={}", dto.getRequestNo());
            throw new BizException(ErrorCode.DUPLICATE_SUBMIT);
        }
        try {
            doSubmit(dto);          // 真正执行业务
            return ApiResult.ok(null);
        } catch (Exception e) {
            // 业务失败必须删掉占坑的 key，否则用户修正参数后重试会被误判为「重复提交」
            redisTemplate.delete(key);
            throw e;
        }
    }

    private void doSubmit(SubmitOrderDTO dto) {
        // 实际下单逻辑：扣库存、插订单……
        log.info("下单成功, requestNo={}, productId={}", dto.getRequestNo(), dto.getProductId());
    }
}

@Mapper
public interface StockMapper {

    /** 乐观锁扣库存：where stock >= quantity 保证「库存够才扣」。返回 0 表示失败，天然防超卖 */
    @Update("UPDATE t_stock SET stock = stock - #{quantity}, version = version + 1 " +
            "WHERE product_id = #{productId} AND stock >= #{quantity}")
    int deduct(@Param("productId") Long productId, @Param("quantity") int quantity);
}

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Idempotent {
    /** 幂等 key 前缀，如 "order" */
    String prefix();
    /** 幂等记录过期时间，单位秒 */
    long expireSeconds() default 600;
}

@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class IdempotentAspect {

    private final StringRedisTemplate redisTemplate;

    @Around("@annotation(idempotent)")
    public Object around(ProceedingJoinPoint joinPoint, Idempotent idempotent) throws Throwable {
        // 从方法参数中提取 requestNo（约定 DTO 里必须有 requestNo 字段）
        String requestNo = extractRequestNo(joinPoint.getArgs());
        if (requestNo == null) {
            throw new BizException(ErrorCode.PARAM_ERROR.getCode(), "幂等接口缺少 requestNo");
        }
        String key = "idempotent:" + idempotent.prefix() + ":" + requestNo;
        Boolean first = redisTemplate.opsForValue()
                .setIfAbsent(key, "1", idempotent.expireSeconds(), TimeUnit.SECONDS);
        if (!Boolean.TRUE.equals(first)) {
            throw new BizException(ErrorCode.DUPLICATE_SUBMIT);
        }
        try {
            return joinPoint.proceed();   // 真正执行被注解的方法
        } catch (Throwable e) {
            redisTemplate.delete(key);    // 业务失败删 key，允许重试
            throw e;
        }
    }

    private String extractRequestNo(Object[] args) {
        for (Object arg : args) {
            if (arg == null) {
                continue;
            }
            try {
                Method getter = arg.getClass().getMethod("getRequestNo");
                Object value = getter.invoke(arg);
                if (value != null) {
                    return value.toString();
                }
            } catch (NoSuchMethodException ignored) {
                // 不是带 requestNo 的 DTO，跳过
            } catch (Exception e) {
                throw new BizException(ErrorCode.SYSTEM_ERROR);
            }
        }
        return null;
    }
}

@Idempotent(prefix = "order", expireSeconds = 600)
public ApiResult<Void> submit(SubmitOrderDTO dto) {
    // 业务逻辑，无需关心幂等
    return ApiResult.ok(null);
}

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderMessageListener {

    private final OrderMapper orderMapper;
    private final StringRedisTemplate redisTemplate;

    @RabbitListener(queues = "order.paid.queue")
    public void onMessage(OrderPaidMessage msg) {
        String dedupKey = "msg:dedup:" + msg.getMessageId();
        // 消费端幂等：用 messageId 占坑，重复投递直接丢弃
        Boolean first = redisTemplate.opsForValue()
                .setIfAbsent(dedupKey, "1", 24, TimeUnit.HOURS);
        if (!Boolean.TRUE.equals(first)) {
            log.info("重复消息被丢弃, messageId={}", msg.getMessageId());
            return;
        }
        try {
            processPaidOrder(msg);
        } catch (Exception e) {
            // 处理失败删掉去重 key，让消息重试时能重新处理
            redisTemplate.delete(dedupKey);
            throw e;
        }
    }

    private void processPaidOrder(OrderPaidMessage msg) {
        log.info("处理订单支付消息, orderNo={}", msg.getOrderNo());
        // 实际业务：把订单状态改为已支付（配合状态机幂等）
    }
}

// 时间戳左移 22 位，占据 62~22 位；机器 ID 左移 12 位，占据 21~12 位；序列号占 11~0 位
long id = (ts << 22) | (workerId << 12) | seq;

public class SimpleSnowflakeIdGenerator {

    /** 起始时间戳：2020-01-01 00:00:00，自定义，用于压缩时间戳位数 */
    private static final long EPOCH = 1577836800000L;
    private static final long WORKER_ID_BITS = 10L;
    private static final long SEQUENCE_BITS = 12L;
    private static final long MAX_WORKER_ID = ~(-1L << WORKER_ID_BITS);   // 1023
    private static final long SEQUENCE_MASK = ~(-1L << SEQUENCE_BITS);     // 4095
    private static final long WORKER_ID_SHIFT = SEQUENCE_BITS;            // 12
    private static final long TIMESTAMP_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS; // 22

    private final long workerId;
    private long lastTimestamp = -1L;
    private long sequence = 0L;

    public SimpleSnowflakeIdGenerator(long workerId) {
        if (workerId < 0 || workerId > MAX_WORKER_ID) {
            throw new IllegalArgumentException("workerId 必须在 0~1023 之间");
        }
        this.workerId = workerId;
    }

    public synchronized long nextId() {
        long timestamp = System.currentTimeMillis();
        if (timestamp < lastTimestamp) {
            // 时钟回拨：当前时间比上次生成时间还小，说明系统时钟被往回拨了
            throw new IllegalStateException("系统时钟回拨，拒绝生成 ID");
        }
        if (timestamp == lastTimestamp) {
            // 同一毫秒：序列号自增，超出 4095 则自旋等待下一毫秒
            sequence = (sequence + 1) & SEQUENCE_MASK;
            if (sequence == 0) {
                timestamp = waitNextMillis(lastTimestamp);
            }
        } else {
            sequence = 0L;   // 新的毫秒，序列号归零
        }
        lastTimestamp = timestamp;
        // 拼接：时间戳左移 22 位 | 机器 ID 左移 12 位 | 序列号
        return ((timestamp - EPOCH) << TIMESTAMP_SHIFT)
                | (workerId << WORKER_ID_SHIFT)
                | sequence;
    }

    /** 自旋等待，直到时钟追上 lastTimestamp */
    private long waitNextMillis(long lastTimestamp) {
        long timestamp = System.currentTimeMillis();
        while (timestamp <= lastTimestamp) {
            timestamp = System.currentTimeMillis();
        }
        return timestamp;
    }
}

@Slf4j
public class RobustSnowflakeIdGenerator {

    private static final long EPOCH = 1577836800000L;
    private static final long WORKER_ID_BITS = 10L;
    private static final long SEQUENCE_BITS = 12L;
    private static final long SEQUENCE_MASK = ~(-1L << SEQUENCE_BITS);
    private static final long WORKER_ID_SHIFT = SEQUENCE_BITS;
    private static final long TIMESTAMP_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS;
    /** 允许的最大回拨毫秒数，超过则判定为异常回拨 */
    private static final long MAX_BACKWARD_MS = 5L;

    private final long workerId;
    private long lastTimestamp = -1L;
    private long sequence = 0L;

    public RobustSnowflakeIdGenerator(long workerId) {
        if (workerId < 0 || workerId > ~(-1L << WORKER_ID_BITS)) {
            throw new IllegalArgumentException("workerId 必须在 0~1023 之间");
        }
        this.workerId = workerId;
    }

    public synchronized long nextId() {
        long timestamp = currentTimeMillis();
        // 时钟回拨处理：小于 5ms 的抖动，自旋等待追平；否则拒绝并告警
        if (timestamp < lastTimestamp) {
            long offset = lastTimestamp - timestamp;
            if (offset <= MAX_BACKWARD_MS) {
                log.warn("检测到时钟回拨 {}ms，自旋等待追平", offset);
                timestamp = waitUntil(lastTimestamp);
            } else {
                log.error("时钟回拨超过阈值 {}ms，拒绝生成 ID，请检查 NTP/系统时钟", offset);
                throw new IllegalStateException("时钟回拨异常，拒绝生成 ID");
            }
        }
        if (timestamp == lastTimestamp) {
            sequence = (sequence + 1) & SEQUENCE_MASK;
            if (sequence == 0) {
                timestamp = waitUntil(lastTimestamp + 1);
            }
        } else {
            sequence = 0L;
        }
        lastTimestamp = timestamp;
        return ((timestamp - EPOCH) << TIMESTAMP_SHIFT)
                | (workerId << WORKER_ID_SHIFT)
                | sequence;
    }

    private long waitUntil(long target) {
        long now = currentTimeMillis();
        while (now < target) {
            now = currentTimeMillis();
        }
        return now;
    }

    /** 单独抽一个方法便于测试时 mock 时间 */
    protected long currentTimeMillis() {
        return System.currentTimeMillis();
    }
}

CREATE TABLE t_id_segment (
    biz      VARCHAR(32) PRIMARY KEY,   -- 业务标识，如 "order"
    max_id   BIGINT NOT NULL,           -- 当前已分配到的最大 ID
    step     INT NOT NULL DEFAULT 1000  -- 每次租的号段大小
) ENGINE=InnoDB;

@Slf4j
@Component
@RequiredArgsConstructor
public class SegmentIdGenerator {

    private final IdSegmentMapper segmentMapper;

    /** 当前正在发放的号段 [start, end] 与当前位置 */
    private volatile long currentStart = 0;
    private volatile long currentEnd = 0;
    private final AtomicLong cursor = new AtomicLong(0);

    public synchronized long nextId(String biz) {
        // 当前号段用尽，去数据库原子地租下一段
        if (cursor.get() >= currentEnd) {
            loadSegment(biz);
        }
        return cursor.incrementAndGet();
    }

    /** 原子地租一段号：UPDATE 让并发线程不会租到重叠的区间 */
    private void loadSegment(String biz) {
        // 原子取号段：UPDATE ... SET max_id = max_id + step WHERE biz = ?
        // 数据库返回更新后的 max_id，则本段区间为 (max_id - step, max_id]
        long newMax = segmentMapper.getAndIncrement(biz);
        long step = segmentMapper.getStep(biz);
        this.currentStart = newMax - step + 1;
        this.currentEnd = newMax;
        this.cursor.set(currentStart - 1);
        log.info("号段已加载, biz={}, range=[{}, {}]", biz, currentStart, currentEnd);
    }
}

@Mapper
public interface IdSegmentMapper {

    /** 原子取号段：同一条 SQL 完成「读 + 加 step」，并发下不会租到重复区间 */
    @Update("UPDATE t_id_segment SET max_id = max_id + step WHERE biz = #{biz}")
    int getAndIncrement(@Param("biz") String biz);

    @Select("SELECT step FROM t_id_segment WHERE biz = #{biz}")
    int getStep(@Param("biz") String biz);
}

@Slf4j
@Component
@RequiredArgsConstructor
public class RedisIdGenerator {

    private final StringRedisTemplate redisTemplate;

    /** Redis INCR 原子自增，天然全局唯一 + 单调递增 */
    public long nextId(String biz) {
        Long id = redisTemplate.opsForValue().increment("id:seq:" + biz);
        return id == null ? 0L : id;
    }
}

@Data
@TableName("t_order")
public class Order {
    // ASSIGN_ID：MyBatis-Plus 内置雪花算法，插入时自动生成趋势递增的全局唯一 ID
    @TableId(type = IdType.ASSIGN_ID)
    private Long id;

    private String orderNo;
    private Long userId;
}

@Slf4j
@Service
@RequiredArgsConstructor
public class ProductCacheService {

    private final ProductMapper productMapper;
    private final StringRedisTemplate redisTemplate;

    private static final String CACHE_KEY = "product:";

    public Product getProduct(Long id) {
        // 1. 先查缓存
        String json = redisTemplate.opsForValue().get(CACHE_KEY + id);
        if (json != null) {
            return JSON.parseObject(json, Product.class);
        }
        // 2. 缓存未命中，查库
        Product product = productMapper.selectById(id);
        if (product == null) {
            return null;
        }
        // 3. 回填缓存（带过期时间，防止 key 永久堆积）
        redisTemplate.opsForValue().set(CACHE_KEY + id, JSON.toJSONString(product), 30, TimeUnit.MINUTES);
        return product;
    }

    public void updateProduct(Product product) {
        // 先更新数据库
        productMapper.updateById(product);
        // 再删除缓存（而不是更新缓存）：下次读时重新加载，避免双写不一致
        redisTemplate.delete(CACHE_KEY + product.getId());
    }
}

@Slf4j
@Service
@RequiredArgsConstructor
public class CacheProtectedService {

    private final StringRedisTemplate redisTemplate;
    private final ProductMapper productMapper;

    private static final String CACHE_KEY = "product:";
    private static final String LOCK_KEY = "lock:rebuild:";

    public Product getProduct(Long id) {
        String key = CACHE_KEY + id;
        String json = redisTemplate.opsForValue().get(key);
        if (json != null) {
            return "NULL".equals(json) ? null : JSON.parseObject(json, Product.class);
        }
        // 缓存未命中，查库并回填
        Product product = productMapper.selectById(id);
        if (product == null) {
            // 防穿透：缓存空值（"NULL"），下次查询直接命中缓存，不打 DB
            redisTemplate.opsForValue().set(key, "NULL", 1, TimeUnit.MINUTES);
            return null;
        }
        // 防雪崩：过期时间加随机值，避免大量 key 同时过期
        long ttl = 30 + ThreadLocalRandom.current().nextInt(10);
        redisTemplate.opsForValue().set(key, JSON.toJSONString(product), ttl, TimeUnit.MINUTES);
        return product;
    }

    /** 防击穿：热点 key 过期时，用分布式锁保证只有一个线程重建缓存，其余等待 */
    public Product getProductWithLock(Long id) {
        String key = CACHE_KEY + id;
        String json = redisTemplate.opsForValue().get(key);
        if (json != null) {
            return JSON.parseObject(json, Product.class);
        }
        // 只有一个线程能拿到重建锁，其他线程自旋重试读缓存
        String lockKey = LOCK_KEY + id;
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 5, TimeUnit.SECONDS);
        if (Boolean.TRUE.equals(locked)) {
            try {
                // 双重检查：拿到锁后再读一次缓存，防止重复查库
                json = redisTemplate.opsForValue().get(key);
                if (json != null) {
                    return JSON.parseObject(json, Product.class);
                }
                Product product = productMapper.selectById(id);
                redisTemplate.opsForValue().set(key, JSON.toJSONString(product), 30, TimeUnit.MINUTES);
                return product;
            } finally {
                redisTemplate.delete(lockKey);   // 重建完释放锁
            }
        } else {
            // 没拿到锁：短暂等待后重试读缓存（生产可用重试几次）
            return retryGetFromCache(key);
        }
    }

    private Product retryGetFromCache(String key) {
        for (int i = 0; i < 3; i++) {
            String json = redisTemplate.opsForValue().get(key);
            if (json != null) {
                return JSON.parseObject(json, Product.class);
            }
            try { Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        }
        return null;
    }
}

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderAsyncService {

    private final RabbitTemplate rabbitTemplate;

    /** 下单异步化：接口只负责落库 + 发消息，耗时动作（发短信、加积分、通知仓库）全部异步 */
    public void createOrderAsync(OrderDTO dto) {
        // 1. 快速落库（这是同步链路里唯一必须同步的部分）
        // orderMapper.insert(...);
        // 2. 发消息，把「下单成功」这个事件广播出去，下游各自消费
        rabbitTemplate.convertAndSend("order.exchange", "order.created", dto);
        log.info("下单消息已发送, orderNo={}", dto.getOrderNo());
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class RedisRateLimiter {

    private final StringRedisTemplate redisTemplate;

    /**
     * 固定窗口限流 Lua 脚本：
     *   KEYS[1] = 限流 key，ARGV[1] = 窗口内允许的最大请求数，ARGV[2] = 窗口秒数
     * 思路：key 不存在则创建并计数 1；存在且未超限则自增；超限则返回 0 拒绝。
     */
    private static final DefaultRedisScript<Long> RATE_LIMIT_SCRIPT = new DefaultRedisScript<>(
            "local current = redis.call('incr', KEYS[1]) " +
            "if current == 1 then " +
            "    redis.call('expire', KEYS[1], ARGV[2]) " +
            "end " +
            "if current > tonumber(ARGV[1]) then " +
            "    return 0 " +
            "end " +
            "return 1",
            Long.class
    );

    /** 尝试获取令牌，true=放行，false=被限流 */
    public boolean tryAcquire(String key, int limit, int windowSeconds) {
        Long result = redisTemplate.execute(RATE_LIMIT_SCRIPT,
                Collections.singletonList(key), String.valueOf(limit), String.valueOf(windowSeconds));
        return result != null && result == 1L;
    }
}

@Slf4j
@Component
public class GuavaRateLimiter {

    /** 每秒允许 100 个请求（令牌桶：固定速率放令牌，允许突发取走桶内令牌） */
    private final RateLimiter rateLimiter = RateLimiter.create(100.0);

    public boolean tryAcquire() {
        // tryAcquire：有令牌立即放行；没令牌立即返回 false（不阻塞等待）
        return rateLimiter.tryAcquire();
    }
}

@RestController
@RequiredArgsConstructor
public class SeckillController {

    private final RedisRateLimiter rateLimiter;
    private final SeckillService seckillService;

    @PostMapping("/seckill")
    public ApiResult<Void> seckill(@RequestParam Long userId, @RequestParam Long productId) {
        // 接口级限流：同一用户每秒最多 5 次，把刷子挡在业务逻辑外面
        if (!rateLimiter.tryAcquire("rate:seckill:" + userId, 5, 1)) {
            throw new BizException(ErrorCode.TOO_MANY_REQUESTS);
        }
        seckillService.seckill(userId, productId);
        return ApiResult.ok(null);
    }
}

@Service
@RequiredArgsConstructor
public class OrderQueryService {

    private final OrderMapper orderMapper;

    // @DS("slave")：该查询路由到从库（需要引入 dynamic-datasource）
    public Order getById(Long id) {
        return orderMapper.selectById(id);
    }
}

spring:
  shardingsphere:
    datasource:
      names: ds0, ds1, ds2, ds3
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/dk_mall_0
        username: root
        password: root
      # ds1 / ds2 / ds3 同理，指向 dk_mall_1 / 2 / 3
    rules:
      sharding:
        tables:
          t_user:
            # 分库 + 分表：库用 user_id % 4，表用 user_id % 8
            actual-data-nodes: ds$->{0..3}.t_user_$->{0..7}
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: db-inline
            table-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: tbl-inline
        sharding-algorithms:
          db-inline:
            type: INLINE
            props:
              algorithm-expression: ds$->{user_id % 4}
          tbl-inline:
            type: INLINE
            props:
              algorithm-expression: t_user_$->{user_id % 8}
    props:
      sql-show: true

// 生成订单号时嵌入分片基因：orderNo 末位 = user_id % 8
public String buildOrderNo(Long userId) {
    long gene = userId % 8;
    // 订单号 = 时间戳 + 雪花序列 + 基因位，末位编码了 user_id 的取模结果
    return System.currentTimeMillis() + String.valueOf(gene);
}

@Configuration
public class ShardingConfig {

    @Bean
    public DataSource dataSource() throws SQLException {
        // 1. 数据源映射：4 个库
        Map<String, DataSource> dataSourceMap = new HashMap<>();
        for (int i = 0; i < 4; i++) {
            dataSourceMap.put("ds" + i, createDataSource("jdbc:mysql://localhost:3306/dk_mall_" + i));
        }

        // 2. 分片规则：t_user 分库（user_id % 4）+ 分表（user_id % 8）
        ShardingRuleConfiguration ruleConfig = new ShardingRuleConfiguration();
        ruleConfig.getTables().add(getUserTableRule());
        // 3. 全局主键策略：雪花算法（对应第 3 章）
        ruleConfig.setKeyGenerators(Collections.singletonMap(
                "snowflake", new AlgorithmConfiguration("SNOWFLAKE", new Properties())));

        Properties props = new Properties();
        props.setProperty("sql-show", "true");
        return ShardingSphereDataSourceFactory.createDataSource(dataSourceMap,
                Collections.singletonList(ruleConfig), props);
    }

    private ShardingTableRuleConfiguration getUserTableRule() {
        ShardingTableRuleConfiguration table = new ShardingTableRuleConfiguration("t_user",
                "ds$->{0..3}.t_user_$->{0..7}");
        // 分库策略：按 user_id 哈希取模
        table.setDatabaseShardingStrategy(new StandardShardingStrategyConfiguration(
                "user_id", "db-inline"));
        // 分表策略：按 user_id 哈希取模
        table.setTableShardingStrategy(new StandardShardingStrategyConfiguration(
                "user_id", "tbl-inline"));
        // 雪花主键
        table.setKeyGenerateStrategy(new KeyGenerateStrategyConfiguration("id", "snowflake"));
        return table;
    }
}

@Slf4j
@Service
@RequiredArgsConstructor
public class OrderTxService {

    private final OrderMapper orderMapper;
    private final OutboxMessageMapper outboxMapper;

    /** 下单 + 写消息表，必须在同一个本地事务里（保证业务与消息原子） */
    @Transactional(rollbackFor = Exception.class)
    public void createOrderWithOutbox(OrderDTO dto) {
        // 1. 业务操作：插入订单
        Order order = toEntity(dto);
        orderMapper.insert(order);

        // 2. 同事务写入本地消息表：如果订单插入失败，这条消息也不会落库
        OutboxMessage msg = new OutboxMessage();
        msg.setTopic("order.created");
        msg.setPayload(JSON.toJSONString(order));
        msg.setStatus(0);   // 0=待发送
        outboxMapper.insert(msg);

        log.info("下单成功并写入消息表, orderNo={}", dto.getOrderNo());
    }
}

CREATE TABLE t_outbox_message (
    id          BIGINT PRIMARY KEY AUTO_INCREMENT,
    topic       VARCHAR(64) NOT NULL,
    payload     TEXT NOT NULL,
    status      TINYINT NOT NULL DEFAULT 0 COMMENT '0=待发送 1=已发送',
    retry_count INT NOT NULL DEFAULT 0,
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    KEY idx_status (status)
) ENGINE=InnoDB;

@Slf4j
@Component
@RequiredArgsConstructor
public class OutboxRelayTask {

    private final OutboxMessageMapper outboxMapper;
    private final RabbitTemplate rabbitTemplate;

    /** 每 5 秒扫一次待发送消息并投递，保证「业务成功后消息最终一定会被发出去」 */
    @Scheduled(fixedDelay = 5000)
    public void relay() {
        List<OutboxMessage> pending = outboxMapper.selectByStatus(0);
        for (OutboxMessage msg : pending) {
            try {
                rabbitTemplate.convertAndSend("order.exchange", msg.getTopic(), msg.getPayload());
                outboxMapper.markSent(msg.getId());   // 投递成功才改状态
            } catch (Exception e) {
                // 投递失败：增加重试次数，下次扫描继续投。达到上限转人工/告警
                log.error("消息投递失败, msgId={}", msg.getId(), e);
                outboxMapper.increaseRetry(msg.getId());
            }
        }
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderCreatedListener {

    private final StringRedisTemplate redisTemplate;
    private final OrderMapper orderMapper;

    @RabbitListener(queues = "order.created.queue")
    public void onOrderCreated(OrderCreatedMessage msg) {
        String dedupKey = "msg:dedup:order:" + msg.getMessageId();
        // 消费端幂等：messageId 占坑，重复投递直接丢弃
        Boolean first = redisTemplate.opsForValue()
                .setIfAbsent(dedupKey, "1", 24, TimeUnit.HOURS);
        if (!Boolean.TRUE.equals(first)) {
            log.info("重复消息被丢弃, messageId={}", msg.getMessageId());
            return;
        }
        try {
            // 状态机：订单只能从「待支付」流转到「已支付」，重复处理天然幂等
            Order order = orderMapper.selectByOrderNo(msg.getOrderNo());
            if (order != null && order.getStatus() == OrderStatus.PAID.getCode()) {
                log.info("订单已支付，忽略重复消息, orderNo={}", msg.getOrderNo());
                return;
            }
            orderMapper.markPaid(msg.getOrderNo());
            log.info("订单状态已更新为已支付, orderNo={}", msg.getOrderNo());
        } catch (Exception e) {
            // 处理失败删去重 key，让 MQ 重投后能重新处理
            redisTemplate.delete(dedupKey);
            throw e;   // 抛异常 = 不 ACK，MQ 会重新投递
        }
    }
}

public interface TccAction {
    /** Try：冻结库存，不真正扣减；幂等（同一笔事务重复 Try 返回成功） */
    boolean tryFreeze(String tccId, Long productId, int quantity);

    /** Confirm：真正扣减库存；幂等 */
    boolean confirm(String tccId);

    /** Cancel：释放冻结的库存；幂等 + 支持空回滚（Try 没执行成功时 Cancel 也要能处理） */
    boolean cancel(String tccId);
}

@Slf4j
@Service
@RequiredArgsConstructor
public class StockTccAction implements TccAction {

    private final StockMapper stockMapper;
    private final FreezeMapper freezeMapper;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean tryFreeze(String tccId, Long productId, int quantity) {
        // 幂等：同一 tccId 的 Try 已执行过，直接返回成功
        if (freezeMapper.existsByTccId(tccId)) {
            return true;
        }
        // 预留资源：把库存从「可用」挪到「冻结」（不真正扣减，只是标记）
        int rows = stockMapper.freeze(productId, quantity);   // where stock >= quantity，stock -= quantity 冻结到冻结表
        if (rows <= 0) {
            return false;   // 预留失败（库存不足）
        }
        freezeMapper.insert(tccId, productId, quantity);      // 记录冻结明细
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean confirm(String tccId) {
        // 幂等：已确认过就跳过
        Freeze freeze = freezeMapper.selectByTccId(tccId);
        if (freeze == null || freeze.getStatus() == FreezeStatus.CONFIRMED.getCode()) {
            return true;
        }
        // 确认：真正扣减（把冻结的库存正式减掉）
        stockMapper.commitFreeze(freeze.getProductId(), freeze.getQuantity());
        freezeMapper.markConfirmed(tccId);
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean cancel(String tccId) {
        // 幂等 + 空回滚：Try 可能没执行成功，Cancel 时冻结记录不存在，直接返回成功
        Freeze freeze = freezeMapper.selectByTccId(tccId);
        if (freeze == null || freeze.getStatus() == FreezeStatus.CANCELLED.getCode()) {
            return true;   // 空回滚：没有可释放的资源
        }
        // 回滚：释放冻结（把冻结的库存退回可用）
        stockMapper.releaseFreeze(freeze.getProductId(), freeze.getQuantity());
        freezeMapper.markCancelled(tccId);
        return true;
    }
}

// Saga 思想伪代码：本地事务 + 补偿动作，由编排器协调
public void sagaOrder() {
    try {
        createOrder();      // T1：创建订单
        deductStock();      // T2：扣库存
        pay();              // T3：支付
    } catch (Exception e) {
        // 逆序补偿：撤销前面已成功的步骤
        cancelPay();        // C3
        restoreStock();     // C2
        cancelOrder();      // C1
    }
}

/** 值对象：金额。不可变、无标识、靠「值」比较相等，封装金额运算规则 */
public final class Money {

    private final BigDecimal amount;
    private final String currency;

    public Money(BigDecimal amount, String currency) {
        if (amount == null || amount.compareTo(BigDecimal.ZERO) < 0) {
            throw new BizException(ErrorCode.PARAM_ERROR.getCode(), "金额不能为空或为负");
        }
        this.amount = amount;
        this.currency = currency;
    }

    public Money add(Money other) {
        if (!this.currency.equals(other.currency)) {
            throw new BizException(ErrorCode.PARAM_ERROR.getCode(), "币种不一致不能相加");
        }
        return new Money(this.amount.add(other.amount), this.currency);
    }

    // 值对象重写 equals/hashCode：按「值」比较，而不是按引用
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof Money other)) return false;
        return amount.compareTo(other.amount) == 0 && currency.equals(other.currency);
    }

    @Override
    public int hashCode() {
        return Objects.hash(amount, currency);
    }
}

/** 领域事件：订单已支付（用过去式命名，表达「已经发生的事实」） */
public record OrderPaidEvent(Long orderId, String orderNo, BigDecimal amount) {
}

// 在聚合根内发布领域事件
public void pay() {
    if (this.status != OrderStatus.PENDING) {
        throw new BizException(ErrorCode.PARAM_ERROR.getCode(), "订单状态不允许支付");
    }
    this.status = OrderStatus.PAID;
    // 发布领域事件：其他上下文（积分、通知）订阅后各自处理
    registerEvent(new OrderPaidEvent(this.id, this.orderNo, this.totalAmount));
}

/**
 * 订单聚合根：封装订单的业务规则。
 * 所有对订单明细、状态的修改都必须通过它，外部不能直接改明细。
 */
public class Order {

    private Long id;
    private String orderNo;
    private Long userId;
    private List<OrderItem> items = new ArrayList<>();
    private BigDecimal totalAmount = BigDecimal.ZERO;
    private OrderStatus status;

    /** 工厂方法：创建订单，封装「订单号生成 + 状态初始化 + 金额计算」规则 */
    public static Order create(Long userId, List<OrderItem> items) {
        Order order = new Order();
        order.userId = userId;
        order.items = new ArrayList<>(items);
        order.status = OrderStatus.PENDING;
        order.orderNo = generateOrderNo();
        order.recalculateAmount();   // 创建时算一次总金额
        return order;
    }

    /** 领域方法：支付。把「已支付」的状态流转规则放在领域对象内部，而不是散落在 Service */
    public void pay() {
        if (this.status != OrderStatus.PENDING) {
            throw new BizException(ErrorCode.PARAM_ERROR.getCode(), "订单状态不允许支付");
        }
        this.status = OrderStatus.PAID;
        // 这里可以发布领域事件 OrderPaid，供其他上下文订阅
    }

    /** 领域方法：添加明细并重算金额，保证「金额永远由明细推导，不会算错」 */
    public void addItem(OrderItem item) {
        if (this.status != OrderStatus.PENDING) {
            throw new BizException(ErrorCode.PARAM_ERROR.getCode(), "订单已提交，不能修改明细");
        }
        this.items.add(item);
        recalculateAmount();
    }

    /** 金额计算是核心业务规则，集中在领域层，任何入口下单都复用同一套算法 */
    private void recalculateAmount() {
        this.totalAmount = items.stream()
                .map(i -> i.getPrice().multiply(BigDecimal.valueOf(i.getQuantity())))
                .reduce(BigDecimal.ZERO, BigDecimal::add);
    }

    // getter 省略；值对象/构造细节见完整项目
}