跨桶 Lua Java 侧调用、补偿机制生产效果与保证金事务权衡

@Component
public class CrossBucketStockOperator {
​
    @Autowired
    private StringRedisTemplate redisTemplate;
​
    /**
     * Lua脚本在服务启动时加载一次，缓存SHA1避免每次传输脚本全文
     */
    private final DefaultRedisScript<String> crossBucketFreezeScript;
​
    public CrossBucketStockOperator() {
        this.crossBucketFreezeScript = new DefaultRedisScript<>();
        crossBucketFreezeScript.setLocation(
            new ClassPathResource("lua/cross_bucket_freeze.lua"));
        crossBucketFreezeScript.setResultType(String.class);
    }
​
    public LuaResponse freezeStock(String activityId, String ticketType,
                                    String userId, int quantity,
                                    int maxPerUser, String orderId) {
        
        // 从配置中心读取桶数（不同活动可能不同）
        int bucketCount = getBucketCount(activityId);
        int startBucket = Math.abs(orderId.hashCode()) % bucketCount;
​
        // ===== 动态构建KEYS =====
        List<String> keys = new ArrayList<>(1 + bucketCount * 2);
        
        // KEYS[1]: 用户限购HSet
        keys.add(String.format("purchased:%s", activityId));
        
        // KEYS[2 .. bucketCount+1]: 各桶总库存
        for (int i = 0; i < bucketCount; i++) {
            keys.add(String.format("stock:%s:%s:%d", 
                activityId, ticketType, i));
        }
        
        // KEYS[bucketCount+2 .. 2*bucketCount+1]: 各桶冻结数
        for (int i = 0; i < bucketCount; i++) {
            keys.add(String.format("frozen:%s:%s:%d", 
                activityId, ticketType, i));
        }
​
        // ===== ARGV =====
        String[] argv = new String[] {
            userId,                          // ARGV[1]
            String.valueOf(quantity),         // ARGV[2]
            String.valueOf(maxPerUser),       // ARGV[3]
            String.valueOf(bucketCount),      // ARGV[4]
            String.valueOf(startBucket),      // ARGV[5]
            activityId                        // ARGV[6] 用于拼活动状态key
        };
​
        // ===== 执行 =====
        String result = redisTemplate.execute(
            crossBucketFreezeScript, keys, argv);
        
        return JSON.parseObject(result, LuaResponse.class);
    }
​
    private int getBucketCount(String activityId) {
        // 从Nacos配置或DB读取，热门活动配10桶，普通活动5桶
        ActivityConfig config = activityConfigCache.get(activityId);
        return config != null ? config.getBucketCount() : 5;
    }
}

@Component
public class LuaScriptPreloader implements ApplicationRunner {
​
    @Autowired
    private StringRedisTemplate redisTemplate;
​
    @Override
    public void run(ApplicationArguments args) {
        String scriptText = loadScriptFromClasspath("lua/cross_bucket_freeze.lua");
        // 对连接池里的连接执行SCRIPT LOAD
        redisTemplate.execute((RedisCallback<String>) connection -> {
            return connection.scriptingCommands()
                .scriptLoad(scriptText.getBytes());
        });
        log.info("Lua scripts preloaded to Redis");
    }
}

// 改造前（可能跨slot）
"stock:act_1001:standard:0"
"frozen:act_1001:standard:3"
​
// 改造后（同一个hash tag，同一slot）
"stock:{act_1001:standard}:0"
"frozen:{act_1001:standard}:3"
"purchased:{act_1001:standard}"

// KEYS构建时统一使用hash tag格式
String hashTag = String.format("{%s:%s}", activityId, ticketType);
keys.add(String.format("purchased:%s", hashTag));
for (int i = 0; i < bucketCount; i++) {
    keys.add(String.format("stock:%s:%d", hashTag, i));
}
for (int i = 0; i < bucketCount; i++) {
    keys.add(String.format("frozen:%s:%d", hashTag, i));
}

Seata全局事务回滚总次数（半年）：约 1,200 次
  ├── 业务正常回滚（库存不足等）：约 1,150 次 → 这些在Lua层就拒绝了，
  │   Redis根本没执行INCRBY，无需补偿
  └── 非业务回滚（超时/异常/TC故障）：约 50 次 → 需要Redis补偿
       ├── TransactionHook 即时补偿成功：46 次（92%）
       ├── stock-log 扫描补偿成功：3 次（6%）
       └── 5分钟全量对账修正：1 次（2%）

// 在各补偿组件里埋的指标
@Component
public class CompensationMetrics {
​
    private final MeterRegistry registry;
​
    // 1. TransactionHook补偿
    public void recordHookCompensation(boolean success) {
        registry.counter("seata_redis_compensation_total",
            "layer", "transaction_hook",
            "result", success ? "success" : "failure"
        ).increment();
    }
​
    // 2. stock-log扫描补偿
    public void recordScanCompensation(int count) {
        registry.counter("seata_redis_compensation_total",
            "layer", "stock_log_scan",
            "result", "success"
        ).increment(count);
        // 同时记录扫描发现的孤儿冻结数
        registry.gauge("stock_orphan_freeze_current", count);
    }
​
    // 3. 全量对账修正
    public void recordReconciliationFix(int count) {
        registry.counter("seata_redis_compensation_total",
            "layer", "reconciliation",
            "result", "success"
        ).increment(count);
    }
​
    // 4. 补偿失败（进入compensation_fail表）
    public void recordCompensationFailure(String orderId, String reason) {
        registry.counter("seata_redis_compensation_failure_total",
            "reason", reason
        ).increment();
    }
​
    // 5. 补偿重试超限
    public void recordRetryExhausted(String orderId) {
        registry.counter("seata_compensation_retry_exhausted_total")
            .increment();
        // 这个指标 > 0 立即告警
    }
}

public DepositFreezeResult freezeDeposit(DepositFreezeRequest request) {
    // 前置检查1：Redis Lua扣库存
    LuaResponse stockResult = crossBucketStockOperator.freezeStock(...);
    if (stockResult.getCode() != 1) {
        return DepositFreezeResult.fail(stockResult.getMsg());
    }
    
    // 前置检查2：钱包余额Redis预检
    String balanceKey = "wallet:available:" + request.getUserId();
    String cachedBalance = redisTemplate.opsForValue().get(balanceKey);
    if (cachedBalance != null) {
        BigDecimal available = new BigDecimal(cachedBalance);
        if (available.compareTo(request.getDepositAmount()) < 0) {
            // 余额不足，需要先回滚已冻结的库存
            crossBucketStockOperator.rollbackFreeze(...);
            return DepositFreezeResult.fail("余额不足");
        }
    }
    // cachedBalance为null时（缓存未命中）不拦截，让Seata事务内的DB校验兜底
    
    // 进入Seata全局事务
    return doFreezeWithSeata(request);
}