本文共 3450 字，大约阅读时间需要 11 分钟。

在我们日常编码过程中，做营销类系统时，一定会遇到发奖、秒杀等业务，在这些业务中，奖品数量的控制尤为重要，如果没控制好，多发了奖品，会对运营成本造成超支，或者影响用户体验。本篇就介绍几种数量控制方案，供大家讨论。

一：通过锁控制数量

最简单直白的方式就是通过java自带的锁来控制数量的发放，伪代码如下：

synchronized(this) {        int count = countByPrizeId(prizeId);        if (count > totalCount) {            throw new BizException("奖品数量不足");        }        fetch(userId, prizeId);    }

这种模式的优点在于：代码简单易读，对于程序员的要求很低。

风险点分析：

1. synchronized只能在单机环境中生效，如果分布式/集群部署，则无法控制数量
2. 程序中形成单点，所有线程都会竞争这个锁，若随着数据量增大，锁内部操作需要时间增长，则会导致大量线程阻塞，最终使服务宕掉。

二：通过数据库控制数量

针对synchronized只能在单机环境中使用的问题，此处考虑将锁放到jvm之外，这里就引入数据库行级锁控制：

比如在奖品表中加入已发放数量，每次通过更新数量，成功则插入用户流水，失败则返回奖品发完，如下：

UPDATE T_PRIZE SET FETCH_COUNT = FETCH_COUNT + 1 WHERE FETCH_COUNT < TOTAL_COUNT;

又或者不增加发放数量字段，通过mysql的insert into select方式能达到一样的效果：

INSERT INTO T_USER_PRIZE SELECT field FROM DUAL WHERE (SELECT COUNT(*) FROM T_USER_PRIZE WHERE T_PRIZE_ID = '1') < '${TOTAL_COUNT}'

这种模式的优点在于：容易理解，且针对业务量小的发奖来说，QPS能达到要求。

风险点分析：

1. 所有压力都堆到了数据库，一旦量超出预期，很容易造成数据库挂掉。

三：通过分布式锁控制数量

通过分布式锁控制数量是一种比较流行的方案，此处用redis做示例，伪代码如下：

// redis锁    private boolean tryLock(String key) {        redisTemplate.execute(connection -> {            while(此处定义获取锁失败重试次数){                if(connection.setNX(key, LocalDateTime.now().getTime() + EXPIRE_TIME)) {                    return true;                }                // 取出时间判断过期了，则新来的key获取锁                Long oldTime = connection.get(key);                if(oldTime > LocalDateTime.now().getTime() + EXPIRE_TIME)) {                    connection.set(key, LocalDateTime.now().getTime() + EXPIRE_TIME);                    return true;                }            }            return false;        })    }    // redis释放锁    private void deleteKey(String key) {        redisTemplate.execute(connection -> connection.del(key);    }    //调用    public void fetch(Long prizeId, String userId) {        // 每个奖品一个锁        String key = REDIS_KEY_PREFIX + prizeId；        try() {            if(tryLock(key)) {                int count = countByPrizeId(prizeId);                if (count > totalCount) {                    throw new BizException("奖品数量不足");                }                fetch(userId, prizeId);            }        } finally {            // 释放锁            deleteKey(key);        }    }

代码中大致写了通过redis实现分布式锁的方式，代码为在markdown中纯手打，如果错误请自行改正。

使用分布式锁有点在于：

1. key分散，每个奖品一个锁，锁竞争变少。（当然，用synchronized也有锁分散方案）。
2. redis操作都在内存中，效率高，不考虑数据库因素，单机能达到近十万级别QPS。

风险点分析：

1. 数据量超过预期，一样会造成程序宕机。
2. redis一旦挂掉或出现通信障碍，整个流程将会不可用（通过集群部署等方案解决）。

四：无锁控制数量

一旦加上锁，效率肯定会变低，这里介绍一种无锁控制数量的方式来实现，同样基于redis来实现，伪代码如下：

// 此方法中简化对redis操作的代码，只写出调用的哪个方法，不再具体写connection那些。    public void fetch(Long prizeId, String userId) {        // 每个奖品一个锁        String key = REDIS_KEY_PREFIX + prizeId；        if (redisTemplate.get(key) == null) {            // 未发奖时统计出数据为0，发奖后统计出具体发放数据            int count = countByPrizeId(prizeId);            // 注意此处使用setNX，不能使用set，否则会有并发问题。            redisTemplate.setNX(key, count);        }        // redis对于key中的值自增，并返回自增后的值        Long count = redisTemplate.incrr(key);        if (count > totalCount) {            throw new BizException("奖品数量不足");        }        fetch(userId, prizeId);    }

此方法原理为使用redis的incrr方法做一个奖品数量的计数器，每个请求进来都会调用自增拿到最新的数量，之后可进行校验。

此方式优点：未使用锁，效率更高。

风险点分析：

1. 同样依赖于redis，redis挂掉会导致程序不可用。
2. 在缓存未命中时去查库的操作，可能会造成同时大批请求进入db，导致db挂掉。可通过预先放入缓存并将缓存的失效时间设置超过活动时间的方式解决。
3. 执行incrr方法之后，若程序抛出异常，可能会造成奖品少发的情况。

以上四种方式，各有优劣，我们可以根据自己的业务进行选择。技术始终是服务于业务的，比如我们做一个发奖活动，用户只有几十个人，那么肯定是选择直接使用synchronized即可，没必要去搞那些分布式锁等等东西。技术方案很多，选择适合我们当前业务场景的即可！

我的博客：www.scarlettbai.com

转载地址：http://brsni.baihongyu.com/