金融系统热点账户问题

@Transactional
public boolean deduct(String accountId, BigDecimal amount) {
    int maxRetry = 3;
    for (int i = 0; i < maxRetry; i++) {
        Account account = accountMapper.selectById(accountId);
        if (account.getBalance().compareTo(amount) < 0) {
            throw new InsufficientBalanceException();
        }
        int rows = accountMapper.updateWithVersion(
            accountId, 
            account.getBalance().subtract(amount), 
            account.getVersion()
        );
        if (rows > 0) return true;  // 更新成功
        // 更新失败，重试
    }
    throw new ConcurrentUpdateException("更新失败，请重试");
}

// 生产者：将扣款请求发送到队列
@Service
public class AccountService {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public String deduct(String accountId, BigDecimal amount, String bizId) {
        DeductMessage msg = new DeductMessage(accountId, amount, bizId);
        // 发送到热点账户专用队列（单分区保证顺序）
        rocketMQTemplate.syncSend("hot-account-topic:tag", msg);
        return "处理中，请稍后查询结果";
    }
}

// 消费者：单线程串行处理
@RocketMQMessageListener(
    topic = "hot-account-topic",
    consumerGroup = "hot-account-consumer",
    consumeMode = ConsumeMode.ORDERLY  // 顺序消费
)
@Service
public class AccountConsumer implements RocketMQListener<DeductMessage> {
    
    @Override
    public void onMessage(DeductMessage msg) {
        // 幂等检查
        if (processedBizIds.contains(msg.getBizId())) return;
        
        // 直接更新，无并发冲突
        accountMapper.deduct(msg.getAccountId(), msg.getAmount());
        processedBizIds.add(msg.getBizId());
    }
}

@Service
public class SplitAccountService {
    
    private static final int SHARD_COUNT = 10;  // 拆分为10个子账户
    
    // 写入：随机选择子账户
    public void deduct(String accountId, BigDecimal amount) {
        int shardIndex = ThreadLocalRandom.current().nextInt(SHARD_COUNT);
        String shardAccountId = accountId + "_" + shardIndex;
        
        // 检查子账户余额是否足够
        Account shard = accountMapper.selectForUpdate(shardAccountId);
        if (shard.getBalance().compareTo(amount) < 0) {
            // 余额不足，尝试其他子账户或触发再平衡
            rebalanceAndDeduct(accountId, amount);
            return;
        }
        accountMapper.deduct(shardAccountId, amount);
    }
    
    // 查询：汇总所有子账户余额
    public BigDecimal getBalance(String accountId) {
        BigDecimal total = BigDecimal.ZERO;
        for (int i = 0; i < SHARD_COUNT; i++) {
            Account shard = accountMapper.selectById(accountId + "_" + i);
            total = total.add(shard.getBalance());
        }
        return total;
    }
    
    // 定期再平衡：将余额均匀分配到各子账户
    @Scheduled(fixedDelay = 60000)
    public void rebalance(String accountId) {
        // 汇总 → 重新均分
    }
}

@Service
public class RedisAccountService {
    
    private static final String BALANCE_KEY = "account:balance:";
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    // 扣款：Redis 原子操作
    public boolean deduct(String accountId, BigDecimal amount) {
        String key = BALANCE_KEY + accountId;
        
        // Lua 脚本保证原子性
        String luaScript = 
            "local balance = tonumber(redis.call('GET', KEYS[1])) " +
            "if balance == nil then return -1 end " +
            "if balance < tonumber(ARGV[1]) then return -2 end " +
            "redis.call('SET', KEYS[1], balance - tonumber(ARGV[1])) " +
            "return 1";
        
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(luaScript, Long.class),
            Collections.singletonList(key),
            amount.toPlainString()
        );
        
        if (result == 1) {
            // 异步写入数据库
            asyncPersist(accountId, amount);
            return true;
        }
        return false;
    }
    
    // 异步持久化到数据库
    @Async
    public void asyncPersist(String accountId, BigDecimal amount) {
        accountMapper.deduct(accountId, amount);
    }
}

// 请求合并：将短时间内的多个请求合并为一个数据库操作
@Service
public class BatchMergeService {
    
    private final BlockingQueue<DeductRequest> queue = new LinkedBlockingQueue<>();
    
    // 接收请求，放入队列
    public CompletableFuture<Boolean> deduct(String accountId, BigDecimal amount) {
        CompletableFuture<Boolean> future = new CompletableFuture<>();
        queue.offer(new DeductRequest(accountId, amount, future));
        return future;
    }
    
    // 定时批量处理（每10ms合并一次）
    @Scheduled(fixedDelay = 10)
    public void batchProcess() {
        List<DeductRequest> batch = new ArrayList<>();
        queue.drainTo(batch, 100);  // 最多取100个
        
        if (batch.isEmpty()) return;
        
        // 合并计算总扣款额
        BigDecimal totalAmount = batch.stream()
            .map(DeductRequest::getAmount)
            .reduce(BigDecimal.ZERO, BigDecimal::add);
        
        // 一次数据库操作
        boolean success = accountMapper.deductBatch(batch.get(0).getAccountId(), totalAmount);
        
        // 通知所有请求结果
        batch.forEach(req -> req.getFuture().complete(success));
    }
}

// Redis 分布式锁
RLock lock = redisson.getLock("account_lock:" + accountId);
try {
    lock.lock(3, TimeUnit.SECONDS);
    // 执行账户操作
    accountMapper.deduct(accountId, amount);
} finally {
    lock.unlock();
}