Redis——由分布式锁造成的重大事故

事故现场

     

          public SeckillActivityRequestVO seckillHandle(SeckillActivityRequestVO request) {
    SeckillActivityRequestVO response;
        String key = "key:" + request.getSeckillId;
        try {
            Boolean lockFlag = redisTemplate.opsForValue().setIfAbsent(key, "val", 10, TimeUnit.SECONDS);
            if (lockFlag) {
                // HTTP请求用户服务进行用户相关的校验
                // 用户活动校验

                // 库存校验
                Object stock = redisTemplate.opsForHash().get(key+":info", "stock");
                assert stock != null;
                if (Integer.parseInt(stock.toString()) <= 0) {
                    // 业务异常
                } else {
                    redisTemplate.opsForHash().increment(key+":info", "stock", -1);
                    // 生成订单
                    // 发布订单创建成功事件
                    // 构建响应VO
                }
            }
        } finally {
            // 释放锁
            stringRedisTemplate.delete("key");
            // 构建响应VO
        }
        return response;
    }

     

事故原因

事故分析

• 没有其他系统风险容错处理  
  由于用户服务吃紧，网关响应延迟，但没有任何应对方式，这是超卖的 导火索 。
• 看似安全的分布式锁其实一点都不安全  
  虽然采用了 set key value [EX seconds] [PX milliseconds][NX|XX] 的方式，但是如果线程A执行的时间较长没有来得及释放，锁就过期了，此时线程B是可以获取到锁的。当线程A执行完成之后，释放锁，实际上就把线程B的锁释放掉了。这个时候，线程C又是可以获取到锁的，而此时如果线程B执行完释放锁实际上就是释放的线程C设置的锁。这是超卖的

• 非原子性的库存校验  
  非原子性的库存校验导致在并发场景下，库存校验的结果不准确。这是超卖的 根本原因 。通过以上分析，问题的根本原因在于库存校验严重依赖了分布式锁。因为在分布式锁正常set、del的情况下，库存校验是没有问题的。但是，当分布式锁不安全可靠的时候，库存校验就没有用了。

解决方案

实现相对安全的分布式锁

     

          public void safedUnLock(String key, String val) {
        String luaScript = "local in = ARGV[1] local curr=redis.call('get', KEYS[1]) if in==curr then redis.call('del', KEYS[1]) end return 'OK'"";
        RedisScript redisScript = RedisScript.of(luaScript);
        redisTemplate.execute(redisScript, Collections.singletonList(key), Collections.singleton(val));
    }

     

实现安全的库存校验

     

          // redis会返回操作之后的结果，这个过程是原子性的
    Long currStock = redisTemplate.opsForHash().increment("key", "stock", -1);

     

改进之后的代码

     

          public SeckillActivityRequestVO seckillHandle(SeckillActivityRequestVO request) {
    SeckillActivityRequestVO response;
        String key = "key:" + request.getSeckillId();
        String val = UUID.randomUUID().toString();
        try {
            Boolean lockFlag = distributedLocker.lock(key, val, 10, TimeUnit.SECONDS);
            if (!lockFlag) {
                // 业务异常
            }

            // 用户活动校验
            // 库存校验，基于redis本身的原子性来保证
            Long currStock = stringRedisTemplate.opsForHash().increment(key + ":info", "stock", -1);
            if (currStock < 0) { // 说明库存已经扣减完了。
                // 业务异常。
                log.error("[抢购下单] 无库存");
            } else {
                // 生成订单
                // 发布订单创建成功事件
                // 构建响应
            }
        } finally {
            distributedLocker.safedUnLock(key, val);
            // 构建响应
        }
        return response;
    }
    复制代码

     

深度思考

分布式锁有必要么

分布式锁的选型

再次思考分布式锁有必要么

     

          // 通过消息提前初始化好，借助ConcurrentHashMap实现高效线程安全
    private static ConcurrentHashMap SECKILL_FLAG_MAP = new ConcurrentHashMap<>();
    // 通过消息提前设置好。由于AtomicInteger本身具备原子性，因此这里可以直接使用HashMap
    private static Map SECKILL_STOCK_MAP = new HashMap<>();

    ...

    public SeckillActivityRequestVO seckillHandle(SeckillActivityRequestVO request) {
    SeckillActivityRequestVO response;

        Long seckillId = request.getSeckillId();
        if(!SECKILL_FLAG_MAP.get(requestseckillId)) {
            // 业务异常
        }
         // 用户活动校验
         // 库存校验
        if(SECKILL_STOCK_MAP.get(seckillId).decrementAndGet() < 0) {
            SECKILL_FLAG_MAP.put(seckillId, false);
            // 业务异常
        }
        // 生成订单
        // 发布订单创建成功事件
        // 构建响应
        return response;
    }

     

总结

特别推荐一个分享架构+算法的优质内容，还没关注的小伙伴，可以长按关注一下：

长按订阅更多精彩▼
如有收获，点个在看，诚挚感谢