基于Redis+Zookeeper+MySQL实现高并发秒杀系统(一)-阿里云开发者社区

为什么使用Redis ： MySQL并发操作，单机最多支撑1000个，了不起了。无论是从性能还是安全来说，Redis的集成都大大解决了系统的并发问题。利用Redis的原子性操作。
为什么使用Zookeeper : 虽然Redis性能非常之高，但是少不了就是应用服务于Redis之间的通信，每一次的通信至少是需要时间的。所以我们应该在应用程序增加本地缓存，但是本地缓存会存在一个问题，在分布式部署下，多台服务器的多个应用程序，缓存不一致，一样会导致秒杀系统Bug（后续会做出介绍）。

单MySQL版本：
一般来说，如果并发性没那么高的话，我们通过以下语句也能做到安全。利用MySQL InnoDB的行级锁。更改库存的时候，使用以下SQL进行更新库存，就可以了。没什么大的问题。但性能非常低。所有的压力全部堆到数据库上。

update set stock = stock - 1 where stock > 0 and id = #{id}

Redis+MySQL：
用Redis做第一层拦截，防止很多无效的请求操作数据库。减少MySQL的压力。记住，Redis能支撑的并发数比MySQL多的多了。
使用Redis的SetNx，自增，自减，Redisson，Lua脚本（我们这里使用自增自减法）。
请求---Redis---MySQL

@PostMapping("/secKill/{productId}")
    public ResultRtn secKill(@PathVariable("productId") Long productId){
        //商品库存的Redis的Key
        String redisKey = "product:"+productId+":stock";
        //Redis库存减一
       Long count =  redisTemplate.opsForValue().decrement(redisKey);
       if(count<0){
           redisTemplate.opsForValue().increment(redisKey);
           return ResultRtn.error("此商品已经售完");
       }
        try{
            productService.updateStock(productId);
        }catch (Exception e){
            //出现异常Redis减的1，在加回来
            redisTemplate.opsForValue().increment(redisKey);
        }
        return ResultRtn.ok("抢购成功");
    }

贴心小课堂：
    大家看代码，可以看到，我们在catch里面加了redis的缓存增加操作。
  这里是为了避免，redis缓存减了之后，后续代码如果出现异常，其实库存应该是不减的。所以在异常的时候，我们需要把缓存减掉的库存在加回来。

JVM缓存+Redis+MySQL：
虽然Redis性能非常之高，但是少不了就是应用服务于Redis之间的通信，每一次的通信至少是需要时间的。并且在我们实际的秒杀场景当中，其实我们的很多请求都算是无效的。
比方说我们某个商品库存只有100个，现在有1000个用户来抢，每人只能抢一个。那其实900个用户都是抢不到的。那么这900个用户的请求其实也没有必要再去与Redis进行请求。
所以我们应该在应用程序增加本地缓存。所谓的本地缓存就是我们的JVM缓存。我们可以在ConcurrentHashMap中存储，标示着我们的商品是否还有库存。如果没有库存，直接返回请求结果“已经被抢完”。
请求---JVM缓存---Redis---数据库。
代码如下：

private static Map<String,Boolean> jvmCache = new ConcurrentHashMap<>();
@PostMapping("/secKill/{productId}/{userId}")
    public ResultRtn secKill(@PathVariable("productId") Long productId,@PathVariable("userId") Long userId){
        //JVM的Key
        String jvmKey = "product"+productId;
        //判断JVM缓存
        if(params.containsKey(jvmKey)){
            return ResultRtn.error("此商品已经售完");
        }
        //商品库存的Redis的Key
        String redisKey = "product:"+productId+":stock";
        //Redis库存减一
       Long count =  redisTemplate.opsForValue().decrement(redisKey);
       if(count<0){
           redisTemplate.opsForValue().increment(redisKey);
           params.put("product"+productId,true);
           return ResultRtn.error("此商品已经售完");
       }
        try{
            productService.updateStock(productId);
        }catch (Exception e){
            params.remove(jvmKey);
            //出现异常Redis减的1，在加回来
            redisTemplate.opsForValue().increment(redisKey);
        }
        return ResultRtn.ok("抢购成功");
    }

贴心小课堂：
    这里也是一样，在catch里面。把两个缓存（JVM与Redis）减去的库存，在加回去。

以上代码就是利用JVM与Redis双重缓存实现秒杀的典型案例。在单服务器部署下，以上代码可以说很安全，几乎应该没有什么大问题了。但是如果我们的应用是在分布式部署的情况下。那JVM缓存多台机器不一致。这个问题就非常严重。场景如下：
比方说，现在有个商品iphone12，库存仅剩最后一个。A线程在A机器执行到代码14行的时候，此时B线程在B机器请求进来，发现库存不足，会将B机器的JVM缓存，商品的Key对应的Value设置为true。接着A机器在执行到代码20行的时候，报错了。那么Redis减的库存应该要加回去，也就说库存还是剩一个，但是B机器的JVM缓存已经标识这个商品被抢光了。那么倘若后面所有的请求或者A机器宕机了，所有的请求到B机器，导致所有人全部抢购失败。到最后库存并没有全部卖出去。出现了少卖现象。
大致执行流程顺序是这样的：
1 ： A请求进入A机器，抢到了最后一个商品，Redis库存扣减为0
2 : 此时此刻A请求还没有完全结束的时候，B请求进来，发现库存不足。则B机器的JVM缓存该商品标识为True了。
3 ：此时A请求在A机器因为某些原因，抛出异常。则A请求刚刚在Redis减掉的库存要加回去。因为抛出异常了，意味着该商品没有抢成功。
4 ：但B机器的JVM缓存已经标识该商品被抢完了。那么如果后续所有的请求全部到B机器，是不是所有的请求都抢不到这个商品了。或者我们说A机器宕机了，所有请求全部到了B机器，所有抢购这个商品的全部都失败。那么这就造成了少卖现象。

解决上述问题，怎么解决呢？就涉及到我们的JVM缓存之间的同步问题。就是当A机器的JVM缓存变动了，B机器或者分布式下的其他机器对应这个缓存，也应该同步刷新。

我们可以通过分布式协调框架Zookeeper来解决这个问题。

前篇代码及问题描述：

private static Map<String,Boolean> params = new ConcurrentHashMap<>();
@PostMapping("/secKill/{productId}/{userId}")
    public ResultRtn secKill(@PathVariable("productId") Long productId,@PathVariable("userId") Long userId){
        //JVM的Key
        String jvmKey = "product"+productId;
         try{
        //判断JVM缓存
        if(params.containsKey(jvmKey)){
            return ResultRtn.error("此商品已经售完");
        }
        //商品库存的Redis的Key
        String redisKey = "product:"+productId+":stock";
        //Redis库存减一
       Long count =  redisTemplate.opsForValue().decrement(redisKey);
       if(count<0){
           redisTemplate.opsForValue().increment(redisKey);
           params.put("product"+productId,true);
           return ResultRtn.error("此商品已经售完");
       }
       
            productService.updateStock(productId);
        }catch (Exception e){
            params.remove(jvmKey);
            //出现异常Redis减的1，在加回来
            redisTemplate.opsForValue().increment(redisKey);
        }
        return ResultRtn.ok("抢购成功");
    }

贴心小课堂：
    这里也是一样，在catch里面。把两个缓存（JVM与Redis）减去的库存，在加回去。

使用分布式协调框架(Zookeeper)解决JVM缓存不一致的问题：
解决思路如下：。

当B机器请求商品为已售完的时候，会将B机器的此商品JVM缓存标识别True，并利用Zookeeper监听此商品节点。
那么在A机器发生异常需要回滚的时候，会利用Zookeeper通知B机器，B机器需要删除此商品的JVM缓存。
源码如下：

配置Zookeeper信息

@Configuration
public class ZookeeperConfig {


    /**
     * zookeeper IP
     */
    @Value("${zookeeper.ip}")
    private String ip;

    /**
     * Zookeeper 端口
     */
    @Value("${zookeeper.host}")
    private String host;

    /**
     * 当前应用的服务端口，用于打印测试用
     */
    @Value("${server.port}")
    private String serverPort;


    @Bean
    public ZooKeeper initZookeeper() throws Exception {
        // 创建观察者
        ZookeeperWatcher watcher = new ZookeeperWatcher(serverPort);
        // 创建 Zookeeper 客户端
        ZooKeeper zooKeeper = new ZooKeeper(ip+":"+host, 30000, watcher);
        // 将客户端注册给观察者
        watcher.setZooKeeper(zooKeeper);
        // 将配置好的 zookeeper 返回
        return zooKeeper;
    }


}

配置Zookeeper监听：

@Service
@Slf4j
public class ZookeeperWatcher  implements Watcher {
    private ZooKeeper zooKeeper;
    public ZooKeeper getZooKeeper() {
        return zooKeeper;
    }
    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }
    private String serverPort;
    public String getServerPort(){
        return this.serverPort;
    }
    public ZookeeperWatcher(){

    }
    public ZookeeperWatcher (String serverPort){
        this.serverPort = serverPort;
    }
    @Override
    public void process(WatchedEvent watchedEvent) {

        System.out.println("************************zookeeper***start*****************");

        if (watchedEvent.getType() == Event.EventType.None && watchedEvent.getPath() == null) {
            log.info("项目启动，初始化zookeeper节点"+getServerPort());
            try {
                // 创建 zookeeper 商品售完信息根节点
                String path = Constants.zoo_product_key_prefix;
                if (zooKeeper != null && zooKeeper.exists(path, false) == null) {
                    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
            } catch (KeeperException | InterruptedException e) {
                e.printStackTrace();
            }

        } else if (watchedEvent.getType() == Event.EventType.NodeDataChanged) {
            try {
                // 获取节点路径
                String path = watchedEvent.getPath();
                // 获取节点数据
                String soldOut = new String(zooKeeper.getData(path, true, new Stat()));
                // 获取商品 Id
                String productId = path.substring(path.lastIndexOf("/") + 1);
                // 处理当前服务器对应 JVM 缓存
                 if("false".equals(soldOut)){
                    log.info("端口:"+getServerPort()+",zookeeper节点："+path+"标识为false【商品并未售完】");
                    if(RedisZookeeperController.getParams().containsKey(Constants.local_product_key_prefix+productId)){
                        RedisZookeeperController.getParams().remove(Constants.local_product_key_prefix+productId);
                    }
                }
            } catch (KeeperException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

秒杀接口修改

 private static ConcurrentHashMap<String,Boolean> params = new ConcurrentHashMap<>();

    @PostMapping("/secKill/{productId}/{userId}")
    public ResultRtn secKill(@PathVariable("productId") Long productId,
                             @PathVariable("userId") Long userId,
                             @RequestParam("isAndEx") String isAndEx
    ) throws KeeperException, InterruptedException {
        //JVM的Key
        String jvmKey = Constants.local_product_key_prefix+productId;
        //判断JVM缓存
        if(params.containsKey(jvmKey)){
            log.info("本地缓存已售完");
            return ResultRtn.ok("此商品已经售完");
        }
        //商品库存的Redis的Key
        String redisKey = Constants.redis_product_key_prefix+productId;
        //Redis库存减一
        Long count =  redisTemplate.opsForValue().decrement(redisKey);
        try{
        /**
         * 【isAndEx】纯属为了测试模拟并发使用，让我们能够看到zookeeper给我们带来的效果
         * 场景如下：
         * 某商品库存为1  启动两个应用  应用1端口：9001   应用2端口：9002
         * A线程在应用1，【isAndEx】传值为1，则A线程在会在此处卡顿10秒，此时Redis库存已为0，10秒倒计时开始
         *【A线程还在卡顿】- B线程在应用2，【isAndEx】传值为为0，则B线程，查询Redis库存为0，返回商品已售完，且应用2的JVM缓存【params】已设置为true
         * A线程在应用1-模拟的10秒结束，抛出异常，减1的库存要加回来。库存变为1，
         * 当因为应用1发生异常，将库存+1的时候，如果没有zookeeper将应用2的JVM缓存【params】删掉，那么后面所有的请求进入到B机器，则全部不会抢到商品。
         * 至此，则会出现少卖现象。
         */
        if("1".equals(isAndEx)){
            Thread.sleep(10000);
            throw new Exception("模拟异常,已延迟5秒");
        }
        if(count<0){
            redisTemplate.opsForValue().increment(redisKey);
            params.put(Constants.local_product_key_prefix+productId,true);
            // zookeeper 中设置售完标记， zookeeper 节点数据格式 product/1 true
            String productPath = Constants.zoo_product_key_prefix + "/" + productId;
            if(zooKeeper.exists(productPath, true)==null){
                zooKeeper.create(productPath, "true".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            return ResultRtn.ok("此商品已经售完");
        }
            productService.updateStock(productId);
        }catch (Exception e){
            // 通过 zookeeper 回滚其他服务器的 JVM 缓存中的商品售完标记
            String path = Constants.zoo_product_key_prefix + "/" + productId;
            if (zooKeeper.exists(path, true) != null){
                zooKeeper.setData(path, "false".getBytes(), -1);
            }
            params.remove(jvmKey);
            //出现异常Redis减的1，在加回来
            redisTemplate.opsForValue().increment(redisKey);
            return ResultRtn.error("网络拥挤，请稍后重试");
        }
        return ResultRtn.ok("抢购成功");
    }


    public static ConcurrentHashMap<String,Boolean> getParams(){
        return params;
    }

测试过程：
为了测试，我们启动两个应用，应用 1（机器A）端口9001，应用2（机器B）端口9002。
为了测试出并发问题，我们在秒杀接口里面增加一个参数，来让9001线程暂停10秒钟。那么9002不进行卡顿。这样我们先调用9001，让库存减1且线程睡眠10秒。那么此时我们访问9002，应该是商品已售完。在9001接口返回结果前，反复调用9002，控制台应打印本地缓存已售完，接口返回此商品已售完，这就说明了9002的JVM缓存已经设置为True。等到9001接口因为10秒后抛出异常，Redis库存应在加回去，那么需要通知到9002，告诉9002，我这个商品没有抢成功，库存加回去了。你（9002）的缓存就不能是已售完的情况。否则如果后续所有请求到达9002，全部抢不成功，但实际上是有库存啊。此时我们在调用9002的秒杀接口，就应该是抢购成功。这样我们Zookeeper的目的就达到了。

利用Idea启动两个应用，复制一个原来的，将端口设置为9002

设置数据库商品库存为1

调用接口，将数据库库存同步到Redis

调用9001秒杀接口, isAndEx传1，Redis库存减之后，将卡顿10秒，给足其他线程秒杀的时间

http://127.0.0.1:9001/redisZookeeper/secKill/1337041728065032193/1?isAndEx=1

9001接口卡顿期间,去调用9002秒杀接口, isAndEx传0

http://127.0.0.1:9001/redisZookeeper/secKill/1337041728065032193/1?isAndEx=0

接口返回：

{
    "code": 0,
    "msg": "此商品已经售完",
    "data": "此商品已经售完"
}

等待9001接口返回结果之后,在去调用9002秒杀接口, isAndEx传0，则抢购成功

http://127.0.0.1:9001/redisZookeeper/secKill/1337041728065032193/1?isAndEx=0

接口返回：

{
    "code": 0,
    "msg": "抢购成功",
    "data": "抢购成功"
}

源码地址： https://gitee.com/stevenlisw/study-project.git

基于Redis+Zookeeper+MySQL实现高并发秒杀系统(一)

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像