扣减机制

Question

org.qiyu.live.bank.provider.service.impl;

jakarta.annotation.;
org.idea.qiyu.live.framework.redis.starter.key.BankProviderCacheKeyBuilder;
org.qiyu.live.bank.dto.AccountTradeReqDTO;
org.qiyu.live.bank.dto.AccountTradeRespDTO;
org.qiyu.live.bank.provider.dao.maper.IQiyuCurrencyAccountMapper;
org.qiyu.live.bank.provider.dao.po.QiyuCurrencyAccountPO;
org.qiyu.live.bank.provider.service.IQiyuCurrencyAccountService;
org.qiyu.live.bank.provider.service.IQiyuCurrencyTradeService;
org.qiyu.live.bank.constants.TradeTypeEnum;
org.slf4j.Logger;
org.slf4j.LoggerFactory;
org.springframework.beans.factory.annotation.;
org.springframework.data.redis.core.RedisTemplate;
org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
org.springframework.stereotype.;
org.springframework.transaction.annotation.;

java.util.concurrent.LinkedBlockingQueue;
java.util.concurrent.ThreadPoolExecutor;
java.util.concurrent.TimeUnit;

QiyuCurrencyAccountServiceImpl IQiyuCurrencyAccountService {

    Logger = LoggerFactory.(QiyuCurrencyAccountServiceImpl.);

    RedisTemplate<String,Object> ;

    BankProviderCacheKeyBuilder ;

    IQiyuCurrencyTradeService ;

    ThreadPoolExecutor = ThreadPoolExecutor(
            ,        ,                   ,                    TimeUnit.,      LinkedBlockingQueue<>() );
    IQiyuCurrencyAccountMapper ;
    ThreadPoolTaskExecutor ;

    (userId) {
        {
            QiyuCurrencyAccountPO accountPO = QiyuCurrencyAccountPO();
            accountPO.setUserId(userId);
            .insert(accountPO);
            ;
        }(Exception e)
        {

        }
        ;
    }

    (userId, num) {
        String cacheKey = .buildUserBalance(userId);
        (.hasKey(cacheKey))
        {
            .opsForValue().increment(cacheKey,num);
            .expire(cacheKey,,TimeUnit.);
        }
        .execute(Runnable() {
            () {
                consumeIncrDBHandler(,);
            }
        });

    }

    (rollbackFor = Exception.)
    (userId, num) {
        .incr(userId,num);
        .insertOne(userId,num, TradeTypeEnum..getCode());
    }

    (userId, num) {
        String cacheKey = .buildUserBalance(userId);
        (.hasKey(cacheKey))
        {
            Long result = .opsForValue().decrement(cacheKey,num);
            .expire(cacheKey,,TimeUnit.);
            result > ;
        }
        .execute(Runnable() {
            () {
                consumeDecrDBHandler(, );
            }
        });
        ;
    }

    (rollbackFor = Exception.)
    (userId, num) {
        .decr(userId, num);
        .insertOne(userId, num * -, TradeTypeEnum..getCode());
    }

    Integer (userId) {
        String cacheKey = .buildUserBalance(userId);
        Object cacheBalance = .opsForValue().get(cacheKey);
        (cacheBalance != )
        {
            ((Integer) cacheBalance == -)
            {
                ;
            }
            (Integer) cacheBalance;
        }
        Integer currentBalance = .queryBalance(userId);
        (currentBalance == )
        {
            .opsForValue().set(cacheKey,-,,TimeUnit.);
            ;
        }
        .opsForValue().set(cacheKey,currentBalance,,TimeUnit.);
        currentBalance;
    }

    AccountTradeRespDTO (AccountTradeReqDTO accountTradeReqDTO) {
        userId = accountTradeReqDTO.getUserId();
        num = accountTradeReqDTO.getNum();
        Integer balance = getBalance(userId);
        (balance == || balance < num)
        {
            AccountTradeRespDTO.(userId,,);
        }
        .decr(userId,num);
        AccountTradeRespDTO.(userId,);
    }

    AccountTradeRespDTO (AccountTradeReqDTO accountTradeReqDTO) {
        AccountTradeRespDTO.(-, );
    }
}

老师 这个钱包的设计，对于redis是否需要考虑原子性，避免并发竞争问题？会不会有重复扣减的情况，现在写的这个代码 这些问题是不是暂时没有考虑进去，在实际工作中，还需要增加哪些工作才能比较完善

Answer 1

Danny_Idea 2025-10-06 11:14:30

最保险的方式是所有钱包扣减行为都走MySQL 然后用事务和行锁做控制，每次扣减前都生成一个唯一的流水id，避免重复扣减行为发生。

• 提问者 光_cfstOQ #1

  那么是不是可以通过MQ的方式，把所有的钱包和扣减行为都交由MQ来异步处理，然后每次更新MySQL了以后更新redis的方式，是不是用这样的方式是最保险的？但如果是这样的方式 如果在高并发场景下  redis是不是会更新延迟？还是说有其他的比较好的解决方式？

• Danny_Idea 回复 提问者 光_cfstOQ #2

  在另一处回答了你这个问题