【系统架构设计百科】阿里巴巴架构：双十一的流量工程

/**
 * 用户 ID 路由到 RZone 的核心逻辑
 * 使用 userId 的后两位作为分片键，将用户映射到 00-99 共 100 个逻辑分片
 * 再通过路由表将逻辑分片映射到物理 RZone
 */
public class UnitRouter {

    // 路由表：逻辑分片 -> 物理 RZone
    // 通过配置中心动态下发，支持运行时切换
    private volatile Map<Integer, String> routeTable;

    /**
     * 根据用户 ID 获取目标 RZone
     */
    public String getTargetUnit(long userId) {
        // 取用户 ID 后两位，得到 0-99 的逻辑分片号
        int slot = (int) (userId % 100);
        // 查路由表获取物理 RZone
        String targetZone = routeTable.get(slot);
        if (targetZone == null) {
            throw new RouteException("No route found for slot: " + slot);
        }
        return targetZone;
    }

    /**
     * 切换指定分片的路由目标
     * 用于单元扩缩容和故障切换
     */
    public void switchRoute(int slot, String newZone) {
        Map<Integer, String> newTable = new HashMap<>(routeTable);
        newTable.put(slot, newZone);
        this.routeTable = Collections.unmodifiableMap(newTable);
    }

    /**
     * 批量切换分片路由
     * 用于整个 RZone 故障时的批量迁移
     */
    public void batchSwitchRoute(String fromZone, String toZone) {
        Map<Integer, String> newTable = new HashMap<>(routeTable);
        for (Map.Entry<Integer, String> entry : newTable.entrySet()) {
            if (entry.getValue().equals(fromZone)) {
                newTable.put(entry.getKey(), toZone);
            }
        }
        this.routeTable = Collections.unmodifiableMap(newTable);
    }
}

# DRC 同步任务配置示例
drc_task:
  name: "gzone_product_sync"
  source:
    cluster: "hz-gzone-mysql"
    database: "product_db"
    tables:
      - name: "product_info"
        columns: ["id", "title", "price", "status", "gmt_modified"]
      - name: "category"
        columns: "*"
  target:
    cluster: "zb-gzone-mysql"
    database: "product_db"
  sync_mode: "bidirectional"
  conflict_strategy: "last_write_wins"
  filter:
    exclude_patterns:
      - "tmp_*"
      - "*_backup"
  monitor:
    delay_threshold_ms: 500
    alert_channels: ["dingding", "sms"]

/**
 * TDDL 分库分表规则配置示例
 * 以订单表为例：按 buyer_id 分 8 个库，每个库 32 张表
 * 总共 256 个分片，支撑千万级 TPS
 */
public class OrderShardingRule implements ShardingRule {

    private static final int DB_COUNT = 8;
    private static final int TABLE_COUNT_PER_DB = 32;

    /**
     * 计算目标库索引
     * 使用 buyer_id 做分库键，保证同一买家的订单落在同一个库
     */
    @Override
    public int calculateDbIndex(long buyerId) {
        // 先对总分片数取模，再除以每库表数
        int totalShardIndex = (int) (buyerId % (DB_COUNT * TABLE_COUNT_PER_DB));
        return totalShardIndex / TABLE_COUNT_PER_DB;
    }

    /**
     * 计算目标表后缀
     */
    @Override
    public int calculateTableIndex(long buyerId) {
        int totalShardIndex = (int) (buyerId % (DB_COUNT * TABLE_COUNT_PER_DB));
        return totalShardIndex % TABLE_COUNT_PER_DB;
    }

    /**
     * 生成物理表名
     * 例如：order_0023 表示第 23 号分片表
     */
    @Override
    public String getPhysicalTableName(long buyerId) {
        int tableIndex = calculateTableIndex(buyerId);
        return String.format("order_%04d", tableIndex);
    }

    /**
     * 生成物理库名
     * 例如：order_db_03 表示第 3 号分片库
     */
    @Override
    public String getPhysicalDbName(long buyerId) {
        int dbIndex = calculateDbIndex(buyerId);
        return String.format("order_db_%02d", dbIndex);
    }
}

/**
 * TDDL SQL 执行流程简化示例
 */
public class TddlExecutor {

    private final SqlParser parser;
    private final ShardingRouter router;
    private final ConnectionPool connPool;

    /**
     * 执行一条 SQL
     */
    public ResultSet execute(String sql, Object[] params) throws SQLException {
        // 阶段一：SQL 解析
        SqlStatement stmt = parser.parse(sql);
        ShardingKey key = stmt.extractShardingKey(params);

        // 阶段二：路由计算
        List<TargetNode> targets = router.route(stmt, key);

        if (targets.size() == 1) {
            // 单分片查询：直接下推
            return executeSingle(targets.get(0), stmt, params);
        }

        // 阶段三：跨分片查询 - 并行执行
        List<Future<ResultSet>> futures = new ArrayList<>();
        for (TargetNode target : targets) {
            futures.add(asyncExecute(target, stmt, params));
        }

        // 阶段四：结果合并
        List<ResultSet> results = collectResults(futures);
        return mergeResults(results, stmt);
    }

    /**
     * 合并多分片结果
     * 处理 ORDER BY、GROUP BY、LIMIT 等聚合操作
     */
    private ResultSet mergeResults(List<ResultSet> results, SqlStatement stmt) {
        if (stmt.hasOrderBy()) {
            return new MergeSortResultSet(results, stmt.getOrderByColumns());
        }
        if (stmt.hasGroupBy()) {
            return new GroupByMergeResultSet(results, stmt.getGroupByColumns());
        }
        return new SimpleUnionResultSet(results);
    }
}

# TDDL 数据源配置示例
tddl:
  app_name: "trade-center"
  group_rules:
    - group: "order_db_00"
      master: "10.0.1.10:3306"
      slaves:
        - host: "10.0.1.11:3306"
          weight: 5
        - host: "10.0.1.12:3306"
          weight: 3
        - host: "10.0.1.13:3306"
          weight: 2
      read_strategy: "weight_random"
      slave_delay_threshold_ms: 1000
      # 从库延迟超过阈值时自动切到主库读
      failover_on_delay: true

    - group: "order_db_01"
      master: "10.0.2.10:3306"
      slaves:
        - host: "10.0.2.11:3306"
          weight: 5
        - host: "10.0.2.12:3306"
          weight: 5
      read_strategy: "round_robin"
      slave_delay_threshold_ms: 800
      failover_on_delay: true

/**
 * 压测流量标记传递机制
 * 通过 ThreadLocal 在整个调用链中传递压测标记
 */
public class ShadowFlag {

    private static final ThreadLocal<Boolean> IS_SHADOW = ThreadLocal.withInitial(() -> false);

    /**
     * 标记当前请求为压测流量
     */
    public static void markAsShadow() {
        IS_SHADOW.set(true);
    }

    /**
     * 清除压测标记
     */
    public static void clear() {
        IS_SHADOW.remove();
    }

    /**
     * 判断当前请求是否为压测流量
     */
    public static boolean isShadow() {
        return IS_SHADOW.get();
    }
}

/**
 * 压测数据路由 - 数据源层面的隔离
 * 压测流量写入影子表（Shadow Table），不影响正式数据
 */
public class ShadowDataSourceRouter {

    private final DataSource normalDataSource;
    private final DataSource shadowDataSource;

    public ShadowDataSourceRouter(DataSource normal, DataSource shadow) {
        this.normalDataSource = normal;
        this.shadowDataSource = shadow;
    }

    /**
     * 根据压测标记选择数据源
     */
    public Connection getConnection() throws SQLException {
        if (ShadowFlag.isShadow()) {
            return shadowDataSource.getConnection();
        }
        return normalDataSource.getConnection();
    }
}

/**
 * 影子表 SQL 改写器
 * 将正式表名改写为影子表名
 * 例如：order -> __shadow_order
 */
public class ShadowTableRewriter {

    private static final String SHADOW_PREFIX = "__shadow_";

    /**
     * 改写 SQL 中的表名为影子表名
     */
    public String rewrite(String sql, List<String> tableNames) {
        if (!ShadowFlag.isShadow()) {
            return sql;
        }
        String rewritten = sql;
        for (String table : tableNames) {
            rewritten = rewritten.replace(table, SHADOW_PREFIX + table);
        }
        return rewritten;
    }
}

"""
压测流量模型生成器
根据历史数据分析，构建符合双十一流量分布的压测模型
"""

import random
import time
from dataclasses import dataclass
from typing import List


@dataclass
class TrafficModel:
    """流量模型定义"""
    qps_curve: List[float]          # 每秒 QPS 曲线
    api_distribution: dict          # API 调用比例
    user_behavior_chain: List[str]  # 用户行为链
    hot_item_ratio: float           # 热点商品比例
    cart_size_distribution: dict    # 购物车大小分布


def build_double11_model() -> TrafficModel:
    """构建双十一零点流量模型"""

    # QPS 曲线：模拟零点前 10 秒到零点后 300 秒的流量变化
    # 零点前缓慢上升，零点瞬间达到峰值，然后缓慢下降
    qps_curve = []
    for t in range(-10, 301):
        if t < 0:
            # 零点前：指数上升
            qps = 50000 * (2 ** (t / 3.0))
        elif t == 0:
            # 零点：峰值
            qps = 583000
        elif t <= 5:
            # 峰值持续 5 秒
            qps = 583000 * (1 - 0.02 * t)
        elif t <= 60:
            # 快速下降
            qps = 500000 * (0.9 ** ((t - 5) / 10.0))
        else:
            # 平稳期
            qps = 200000 + random.gauss(0, 5000)
        qps_curve.append(max(0, qps))

    # API 调用比例：基于历史数据
    api_distribution = {
        "query_item": 0.35,          # 商品查询
        "add_cart": 0.15,            # 加购物车
        "create_order": 0.20,        # 下单
        "pay_order": 0.15,           # 支付
        "query_order": 0.10,         # 查询订单
        "query_logistics": 0.05,     # 查物流
    }

    # 用户行为链：典型的购买路径
    user_behavior_chain = [
        "browse_item",
        "add_to_cart",
        "checkout",
        "confirm_order",
        "pay",
        "query_order_status",
    ]

    return TrafficModel(
        qps_curve=qps_curve,
        api_distribution=api_distribution,
        user_behavior_chain=user_behavior_chain,
        hot_item_ratio=0.2,
        cart_size_distribution={1: 0.3, 2: 0.25, 3: 0.2, 5: 0.15, 10: 0.1},
    )


def generate_pressure_plan(model: TrafficModel, target_peak_qps: int) -> dict:
    """生成压测计划"""
    scale_factor = target_peak_qps / max(model.qps_curve)
    scaled_curve = [int(qps * scale_factor) for qps in model.qps_curve]

    return {
        "total_duration_sec": len(scaled_curve),
        "peak_qps": target_peak_qps,
        "qps_curve": scaled_curve,
        "api_distribution": model.api_distribution,
        "estimated_total_requests": sum(scaled_curve),
    }

-- 影子表数据清理脚本
-- 在压测结束后执行

-- 清理影子订单表
TRUNCATE TABLE __shadow_order;

-- 清理影子支付表
TRUNCATE TABLE __shadow_payment;

-- 清理影子库存变更日志
DELETE FROM __shadow_inventory_log
WHERE gmt_create >= '2026-10-20 00:00:00'
  AND gmt_create <= '2026-10-21 00:00:00';

-- 重置影子序列号
ALTER TABLE __shadow_order AUTO_INCREMENT = 1;

-- 验证清理结果
SELECT '__shadow_order' AS table_name, COUNT(*) AS remaining_rows
FROM __shadow_order
UNION ALL
SELECT '__shadow_payment', COUNT(*)
FROM __shadow_payment
UNION ALL
SELECT '__shadow_inventory_log', COUNT(*)
FROM __shadow_inventory_log;

/**
 * 弹性伸缩决策引擎
 * 根据实时指标和预测模型决定扩缩容
 */
public class AutoScalingEngine {

    // 扩容触发阈值
    private static final double CPU_SCALE_UP_THRESHOLD = 0.70;
    // 缩容触发阈值
    private static final double CPU_SCALE_DOWN_THRESHOLD = 0.30;
    // 最小观测窗口（秒）
    private static final int OBSERVATION_WINDOW = 60;
    // 冷却时间（秒），避免频繁伸缩
    private static final int COOLDOWN_PERIOD = 300;

    /**
     * 计算目标实例数
     */
    public ScalingDecision evaluate(ClusterMetrics metrics) {
        double avgCpu = metrics.getAvgCpuUsage();
        double avgMemory = metrics.getAvgMemoryUsage();
        int currentInstances = metrics.getCurrentInstanceCount();
        double avgRt = metrics.getAvgResponseTimeMs();

        // 多维度综合判断
        if (avgCpu > CPU_SCALE_UP_THRESHOLD || avgRt > 200) {
            // 计算需要扩容到多少实例
            double targetCpuUsage = 0.50;
            int targetInstances = (int) Math.ceil(
                currentInstances * avgCpu / targetCpuUsage
            );
            // 单次扩容不超过当前实例数的 50%
            int maxScaleUp = (int) (currentInstances * 1.5);
            targetInstances = Math.min(targetInstances, maxScaleUp);

            return new ScalingDecision(
                ScalingAction.SCALE_UP,
                targetInstances,
                String.format("CPU=%.1f%%, RT=%.0fms", avgCpu * 100, avgRt)
            );
        }

        if (avgCpu < CPU_SCALE_DOWN_THRESHOLD && avgRt < 50) {
            int targetInstances = (int) Math.ceil(
                currentInstances * avgCpu / 0.50
            );
            // 至少保留最小实例数
            targetInstances = Math.max(targetInstances, getMinInstances());

            return new ScalingDecision(
                ScalingAction.SCALE_DOWN,
                targetInstances,
                String.format("CPU=%.1f%%, RT=%.0fms", avgCpu * 100, avgRt)
            );
        }

        return new ScalingDecision(ScalingAction.NO_OP, currentInstances, "stable");
    }

    private int getMinInstances() {
        return 3;
    }
}

/**
 * Sentinel 限流规则配置示例
 * 实现令牌桶 + 滑动窗口的多层限流
 */
public class SentinelConfig {

    /**
     * 配置交易接口的限流规则
     */
    public static void initFlowRules() {
        List<FlowRule> rules = new ArrayList<>();

        // 第一层：接口级别限流（集群维度）
        FlowRule clusterRule = new FlowRule();
        clusterRule.setResource("createOrder");
        clusterRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        clusterRule.setCount(100000);
        clusterRule.setClusterMode(true);
        clusterRule.setClusterConfig(
            new ClusterFlowConfig()
                .setFlowId(1001L)
                .setThresholdType(ClusterRuleConstant.FLOW_THRESHOLD_GLOBAL)
        );
        rules.add(clusterRule);

        // 第二层：单机级别限流
        FlowRule localRule = new FlowRule();
        localRule.setResource("createOrder");
        localRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        localRule.setCount(500);
        localRule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_WARM_UP);
        localRule.setWarmUpPeriodSec(10);
        rules.add(localRule);

        // 第三层：热点参数限流（防止单商品被刷）
        // 限制单个商品 ID 的下单 QPS
        ParamFlowRule hotRule = new ParamFlowRule();
        hotRule.setResource("createOrder");
        hotRule.setParamIdx(0);
        hotRule.setCount(50);
        hotRule.setDurationInSec(1);

        FlowRuleManager.loadRules(rules);
    }
}

/**
 * 预案执行引擎
 * 预案是一组原子操作的有序编排
 */
public class ContingencyPlanEngine {

    private final Map<String, ContingencyPlan> planRegistry = new ConcurrentHashMap<>();

    /**
     * 注册预案
     */
    public void register(ContingencyPlan plan) {
        planRegistry.put(plan.getId(), plan);
    }

    /**
     * 执行预案
     */
    public PlanResult execute(String planId, Map<String, Object> context) {
        ContingencyPlan plan = planRegistry.get(planId);
        if (plan == null) {
            return PlanResult.notFound(planId);
        }

        List<ActionResult> actionResults = new ArrayList<>();
        for (PlanAction action : plan.getActions()) {
            try {
                ActionResult result = action.execute(context);
                actionResults.add(result);

                if (!result.isSuccess() && action.isRequired()) {
                    // 必要步骤失败，中止预案并回滚已执行步骤
                    rollback(actionResults);
                    return PlanResult.failed(planId, actionResults);
                }
            } catch (Exception e) {
                actionResults.add(ActionResult.error(action.getName(), e));
                if (action.isRequired()) {
                    rollback(actionResults);
                    return PlanResult.failed(planId, actionResults);
                }
            }
        }
        return PlanResult.success(planId, actionResults);
    }

    /**
     * 回滚已执行的操作
     */
    private void rollback(List<ActionResult> results) {
        // 逆序回滚
        for (int i = results.size() - 1; i >= 0; i--) {
            ActionResult result = results.get(i);
            if (result.isSuccess() && result.getRollbackAction() != null) {
                try {
                    result.getRollbackAction().execute(Collections.emptyMap());
                } catch (Exception e) {
                    // 回滚失败记录告警，人工介入
                    log.error("Rollback failed for action: {}", result.getActionName(), e);
                }
            }
        }
    }
}

/**
 * RocketMQ 事务消息使用示例
 * 场景：下单时需要同时扣减库存，保证订单和库存的一致性
 */
public class OrderTransactionProducer {

    private final TransactionMQProducer producer;

    public OrderTransactionProducer() {
        this.producer = new TransactionMQProducer("order_transaction_group");
        this.producer.setNamesrvAddr("namesrv1:9876;namesrv2:9876");
        this.producer.setTransactionListener(new OrderTransactionListener());
    }

    /**
     * 发送事务消息
     * 流程：
     * 1. 发送 Half 消息到 Broker
     * 2. Broker 返回成功后，执行本地事务（创建订单）
     * 3. 根据本地事务结果，提交或回滚消息
     * 4. 消费者收到消息后扣减库存
     */
    public SendResult createOrderWithTransaction(OrderDTO order) throws Exception {
        Message msg = new Message(
            "ORDER_TOPIC",
            "CREATE",
            order.getOrderId().getBytes(),
            JSON.toJSONBytes(order)
        );
        // 将订单信息作为参数传递给本地事务执行器
        return producer.sendMessageInTransaction(msg, order);
    }
}

/**
 * 事务消息监听器
 */
public class OrderTransactionListener implements TransactionListener {

    private final OrderService orderService;
    private final TransactionLogMapper txLogMapper;

    /**
     * 执行本地事务
     * 在 Half 消息发送成功后被回调
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        OrderDTO order = (OrderDTO) arg;
        try {
            // 执行本地事务：创建订单 + 记录事务日志
            orderService.createOrder(order);
            txLogMapper.insert(new TransactionLog(
                order.getOrderId(),
                "ORDER_CREATE",
                "SUCCESS"
            ));
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    /**
     * 事务回查
     * 当 Broker 长时间未收到提交/回滚指令时回调
     * 通过查询事务日志判断本地事务是否执行成功
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String orderId = new String(msg.getKeys().getBytes());
        TransactionLog log = txLogMapper.findByOrderId(orderId);
        if (log != null && "SUCCESS".equals(log.getStatus())) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

/**
 * TCC 分布式事务示例：转账场景
 * Try：冻结金额
 * Confirm：扣减冻结金额，增加对方余额
 * Cancel：解冻金额
 */
public interface AccountTccService {

    /**
     * Try 阶段：冻结转出金额
     * 在 account 表中将 amount 从 balance 转移到 frozen
     */
    @TwoPhaseBusinessAction(
        name = "accountTransfer",
        commitMethod = "confirm",
        rollbackMethod = "cancel"
    )
    boolean tryFreezeAmount(
        BusinessActionContext context,
        @BusinessActionContextParameter(paramName = "accountId") String accountId,
        @BusinessActionContextParameter(paramName = "amount") BigDecimal amount
    );

    /**
     * Confirm 阶段：确认扣减
     * 将 frozen 中的金额正式扣除
     */
    boolean confirm(BusinessActionContext context);

    /**
     * Cancel 阶段：取消冻结
     * 将 frozen 中的金额还原到 balance
     */
    boolean cancel(BusinessActionContext context);
}

/**
 * TCC 服务实现
 */
public class AccountTccServiceImpl implements AccountTccService {

    private final AccountMapper accountMapper;
    private final TccTransactionLogMapper logMapper;

    @Override
    public boolean tryFreezeAmount(
        BusinessActionContext context, String accountId, BigDecimal amount
    ) {
        // 幂等检查：防止重复冻结
        String xid = context.getXid();
        if (logMapper.existsByXid(xid)) {
            return true;
        }

        // 检查余额是否充足
        Account account = accountMapper.selectForUpdate(accountId);
        if (account.getBalance().compareTo(amount) < 0) {
            throw new InsufficientBalanceException(accountId, amount);
        }

        // 冻结金额：balance 减少，frozen 增加
        accountMapper.freezeAmount(accountId, amount);

        // 记录事务日志
        logMapper.insert(new TccTransactionLog(xid, accountId, amount, "TRY"));
        return true;
    }

    @Override
    public boolean confirm(BusinessActionContext context) {
        String xid = context.getXid();
        String accountId = (String) context.getActionContext("accountId");
        BigDecimal amount = new BigDecimal(
            context.getActionContext("amount").toString()
        );

        // 幂等检查
        TccTransactionLog log = logMapper.findByXid(xid);
        if (log != null && "CONFIRMED".equals(log.getStatus())) {
            return true;
        }

        // 确认扣减：减少 frozen
        accountMapper.confirmDeduction(accountId, amount);

        // 更新事务日志
        logMapper.updateStatus(xid, "CONFIRMED");
        return true;
    }

    @Override
    public boolean cancel(BusinessActionContext context) {
        String xid = context.getXid();
        String accountId = (String) context.getActionContext("accountId");
        BigDecimal amount = new BigDecimal(
            context.getActionContext("amount").toString()
        );

        // 幂等检查
        TccTransactionLog log = logMapper.findByXid(xid);
        if (log == null) {
            // 空回滚：Try 未执行，直接记录日志防悬挂
            logMapper.insert(new TccTransactionLog(xid, accountId, amount, "CANCELLED"));
            return true;
        }
        if ("CANCELLED".equals(log.getStatus())) {
            return true;
        }

        // 取消冻结：frozen 减少，balance 增加
        accountMapper.unfreezeAmount(accountId, amount);

        // 更新事务日志
        logMapper.updateStatus(xid, "CANCELLED");
        return true;
    }
}

/**
 * Seata AT 模式使用示例
 * 只需在业务方法上添加 @GlobalTransactional 注解
 * Seata 自动完成分支事务注册、提交和回滚
 */
@Service
public class OrderServiceImpl implements OrderService {

    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private InventoryFeignClient inventoryClient;
    @Autowired
    private AccountFeignClient accountClient;

    /**
     * 创建订单的全局事务
     * Seata 在 SQL 执行前后自动记录 undo_log
     * 如果任何分支失败，自动回滚所有分支
     */
    @GlobalTransactional(name = "create-order", timeoutMills = 30000)
    @Override
    public OrderResult createOrder(OrderRequest request) {
        // 分支一：创建订单记录
        Order order = new Order();
        order.setOrderId(IdGenerator.nextId());
        order.setBuyerId(request.getBuyerId());
        order.setItemId(request.getItemId());
        order.setQuantity(request.getQuantity());
        order.setTotalAmount(request.getTotalAmount());
        order.setStatus("CREATED");
        orderMapper.insert(order);

        // 分支二：扣减库存（远程调用，Seata 自动传播事务上下文）
        inventoryClient.deduct(request.getItemId(), request.getQuantity());

        // 分支三：扣减账户余额
        accountClient.debit(request.getBuyerId(), request.getTotalAmount());

        return new OrderResult(order.getOrderId(), "SUCCESS");
    }
}

/**
 * 库存预分桶方案
 * 将一件商品的 10000 件库存分散到 8 个 RZone
 */
public class InventoryBucketService {

    private static final int BUCKET_COUNT = 8;

    /**
     * 初始化库存分桶
     * 将总库存均匀分配到各个桶
     */
    public void initBuckets(String itemId, int totalStock) {
        int perBucket = totalStock / BUCKET_COUNT;
        int remainder = totalStock % BUCKET_COUNT;

        for (int i = 0; i < BUCKET_COUNT; i++) {
            int bucketStock = perBucket + (i < remainder ? 1 : 0);
            inventoryMapper.insertBucket(new InventoryBucket(
                itemId,
                i,
                bucketStock,
                bucketStock
            ));
        }
    }

    /**
     * 扣减库存
     * 先尝试本地桶，不足时向其他桶借调
     */
    public boolean deductStock(String itemId, int quantity, int localBucketId) {
        // 尝试本地桶扣减
        int affected = inventoryMapper.deductBucket(itemId, localBucketId, quantity);
        if (affected > 0) {
            return true;
        }

        // 本地桶不足，尝试从其他桶借调
        for (int i = 0; i < BUCKET_COUNT; i++) {
            if (i == localBucketId) {
                continue;
            }
            affected = inventoryMapper.deductBucket(itemId, i, quantity);
            if (affected > 0) {
                return true;
            }
        }

        return false;
    }
}

/**
 * 异步下单队列
 * 用户点击"立即购买"后进入排队
 * 后端按照数据库写入能力匀速处理
 */
public class AsyncOrderProcessor {

    // 内存队列容量
    private static final int QUEUE_CAPACITY = 100000;
    private final BlockingQueue<OrderRequest> orderQueue =
        new LinkedBlockingQueue<>(QUEUE_CAPACITY);
    // 消费线程池
    private final ExecutorService executor =
        Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);

    /**
     * 接收下单请求，放入队列
     * 如果队列已满，直接返回"系统繁忙"
     */
    public OrderResponse submitOrder(OrderRequest request) {
        boolean offered = orderQueue.offer(request);
        if (!offered) {
            return OrderResponse.busy("系统繁忙，请稍后重试");
        }
        return OrderResponse.queued(request.getRequestId());
    }

    /**
     * 消费线程：持续从队列取出请求并处理
     */
    public void startConsumers() {
        for (int i = 0; i < executor.toString().length(); i++) {
            executor.submit(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        OrderRequest request = orderQueue.poll(100, TimeUnit.MILLISECONDS);
                        if (request != null) {
                            processOrder(request);
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            });
        }
    }

    private void processOrder(OrderRequest request) {
        try {
            OrderResult result = orderService.createOrder(request);
            // 通过长连接推送结果给前端
            pushService.pushOrderResult(request.getUserId(), result);
        } catch (Exception e) {
            pushService.pushOrderResult(
                request.getUserId(),
                OrderResult.failed(request.getRequestId(), e.getMessage())
            );
        }
    }
}

/**
 * 分桶计数器实现
 * 适用于高并发场景下的计数需求（如库存、秒杀）
 * 将单一计数器拆分为多个桶，降低 CAS 竞争
 */
public class BucketCounter {

    private final AtomicLong[] buckets;
    private final int bucketCount;

    public BucketCounter(int bucketCount) {
        this.bucketCount = bucketCount;
        this.buckets = new AtomicLong[bucketCount];
        for (int i = 0; i < bucketCount; i++) {
            buckets[i] = new AtomicLong(0);
        }
    }

    /**
     * 递增计数
     * 使用线程 ID 哈希选择桶，分散竞争
     */
    public void increment() {
        int index = (int) (Thread.currentThread().getId() % bucketCount);
        buckets[index].incrementAndGet();
    }

    /**
     * 递减计数
     */
    public boolean decrement() {
        int index = (int) (Thread.currentThread().getId() % bucketCount);
        long current = buckets[index].get();
        if (current > 0) {
            return buckets[index].compareAndSet(current, current - 1);
        }
        // 本桶不足，尝试其他桶
        for (int i = 0; i < bucketCount; i++) {
            if (i == index) continue;
            current = buckets[i].get();
            if (current > 0 && buckets[i].compareAndSet(current, current - 1)) {
                return true;
            }
        }
        return false;
    }

    /**
     * 获取总计数
     */
    public long getTotal() {
        long total = 0;
        for (AtomicLong bucket : buckets) {
            total += bucket.get();
        }
        return total;
    }

    /**
     * 设置总计数（均匀分配到各桶）
     */
    public void setTotal(long total) {
        long perBucket = total / bucketCount;
        long remainder = total % bucketCount;
        for (int i = 0; i < bucketCount; i++) {
            buckets[i].set(perBucket + (i < remainder ? 1 : 0));
        }
    }
}