黑马点评(四) -- 分布式锁

1 . 分布式锁基本原理和实现方式对比

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。

分布式锁的核心思想就是让大家都使用同一把锁，只要大家使用的是同一把锁，那么我们就能锁住线程，不让线程进行，让程序串行执行，这就是分布式锁的核心思路 :

那么分布式锁他应该满足一些什么样的条件呢？

可见性：多个线程都能看到相同的结果，注意：这个地方说的可见性并不是并发编程中指的内存可见性，只是说多个进程之间都能感知到变化的意思

互斥：互斥是分布式锁的最基本的条件，使得程序串行执行

高可用：程序不易崩溃，时时刻刻都保证较高的可用性

高性能：由于加锁本身就让性能降低，所有对于分布式锁本身需要他就较高的加锁性能和释放锁性能

安全性：安全也是程序中必不可少的一环

常见的分布式锁有三种

Mysql：mysql本身就带有锁机制，但是由于mysql性能本身一般，所以采用分布式锁的情况下，其实使用mysql作为分布式锁比较少见

Redis：redis作为分布式锁是非常常见的一种使用方式，现在企业级开发中基本都使用redis或者zookeeper作为分布式锁，利用setnx这个方法，如果插入key成功，则表示获得到了锁，如果有人插入成功，其他人插入失败则表示无法获得到锁，利用这套逻辑来实现分布式锁

Zookeeper：zookeeper也是企业级开发中较好的一个实现分布式锁的方案，由于本套视频并不讲解zookeeper的原理和分布式锁的实现，所以不过多阐述

对于redis的setnx只有当数据不存在的时候才能够set成功 , 为了防止服务出现故障而出现锁不释放，可以给setnx设置一个过期时间 ;

2 . Redis分布式锁的实现核心思路

实现分布式锁时需要实现的两个基本方法：

获取锁：
- 互斥：确保只能有一个线程获取锁
- 非阻塞：尝试一次，成功返回true，失败返回false
释放锁：
- 手动释放
- 超时释放：获取锁时添加一个超时时间

获取锁 :

但是先setnx然后再设置过期时间，可能会出现设置过期时间的时候已经宕机，所以两条合为一条 :

set lock thread1 EX 10 NX

释放锁 :

直接删除key即可 :

整体逻辑 :

3 . 实现分布式锁版本一

先实现锁的基本接口 :

1 . 实现分布式锁接口 :

package com.hmdp.utils;

import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

public class SimpleRedisLock implements ILock{

    private String name ; // 业务名称
    private StringRedisTemplate stringRedisTemplate ;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
}

    private static final String KEY_PREFIX = "lock:" ; // 锁的前缀

    /**
     * 尝试获取锁
     * @param timeoutSec : 锁持有的超时时间 ， 过期后自动释放 ;
     * @return true代表获取锁成功 ， false代表获取锁失败 ;
     */
    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取当前线程id :
        long threadId = Thread.currentThread().getId() ;
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name  , threadId + "", timeoutSec, TimeUnit.SECONDS) ;
        return Boolean.TRUE.equals(success);
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        // 释放锁
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
}

2 . 使用分布式锁 :

将之前的synchronized注释掉，改成自己实现的分布式锁;

整个impl完整代码 :

package com.hmdp.service.impl;

import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.SimpleRedisLock;
import com.hmdp.utils.UserHolder;
import org.springframework.aop.framework.AopContext;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.time.LocalDateTime;


@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private RedisIdWorker redisIdWorker ; // 注入id生成器

    @Resource
    private ISeckillVoucherService seckillVoucherService ;

    @Resource
    private StringRedisTemplate stringRedisTemplate ;

    /**
     * 优惠卷秒杀下单
     * @param voucherId
     * @return
     */
    @Override
    // @Transactional
    public Result seckillVoucher(Long voucherId) {
        // 1 . 查询优惠卷
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId) ;

        // 2 . 判断秒杀是否开始
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始") ;
        }
        // 3 . 判断秒杀是否已经结束
        if(voucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀已经结束") ;
        }
        // 4 . 判断库存是否充足
        if(voucher.getStock()<1){
            return Result.fail("库存不足") ;
        }

        Long userId = UserHolder.getUser().getId();

        // 创建锁对象
        SimpleRedisLock lock = new SimpleRedisLock("order:"+userId,stringRedisTemplate);
        // 获取锁
        boolean isLock = lock.tryLock(1200);
        // 判断是否获取锁成功
        if(!isLock) {
            // 获取锁失败，返回错误或重试
            return Result.fail("不允许重复下单") ;
        }

        try{
            // 获取代理对象(事务)
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy() ;
            return proxy.createVoucherOrder(voucherId);
        }finally {
            // 释放锁
            lock.unlock();
        }

//        synchronized (userId.toString().intern()) {//给每一个用户加一把锁
//            // 获取代理对象(事务)
//            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy() ;
//            return proxy.createVoucherOrder(voucherId);
//        }

    }

    /**
     * 加锁
     * @param voucherId
     * @return
     */
    @Transactional
    public  Result createVoucherOrder(Long voucherId) {
        // 5 . 一人一单
        Long userId = UserHolder.getUser().getId();
        // 5 . 1 查询订单
        int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        // 5 . 2 判断用户师是否存在
        if (count > 0) {
            // 用户已经购买过了
            return Result.fail("用户已经购买了一次!");
        }
        // 6 . 扣减库存
        boolean tag = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherId)
                .gt("stock", 0)
                .update();
        if (!tag) {
            return Result.fail("库存不足");
        }
        // 6 . 创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 6 . 1 订单id
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 6 . 2 用户id
        // Long userId = UserHolder.getUser().getId();
        voucherOrder.setUserId(userId);
        // 6 . 3 代金卷id
        voucherOrder.setVoucherId(voucherId);
        save(voucherOrder);

        // 7 . 返回订单id
        return Result.ok(orderId);

    }
}

3 . 测试 :

在postman中同样用两个request来模拟同一个用户下单 :

可以发现8082获取锁成功， 8081获取锁失败 :

数据库中也只产生了一条数据;

4 . Redis分布式锁误删情况说明

持有锁的线程在锁的内部出现了阻塞，导致他的锁自动释放，这时其他线程，线程2来尝试获得锁，就拿到了这把锁，然后线程2在持有锁执行过程中，线程1反应过来，继续执行，而线程1执行过程中，走到了删除锁逻辑，此时就会把本应该属于线程2的锁进行删除，这就是误删别人锁的情况说明

解决方案：解决方案就是在每个线程释放锁的时候，去判断一下当前这把锁是否属于自己，如果属于自己，则不进行锁的删除，假设还是上边的情况，线程1卡顿，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1反应过来，然后删除锁，但是线程1，一看当前这把锁不是属于自己，于是不进行删除锁逻辑，当线程2走到删除锁逻辑时，如果没有卡过自动释放锁的时间点，则判断当前这把锁是属于自己的，于是删除这把锁。

也就是在释放锁的时候，查看一下当前锁标识是否是自己，如果是，再释放锁，否则，直接跳过;

5 . 解决Redis分布式锁误删问题

之前是直接用线程id当作标识，但是在jvm中线程id是递增的，所以在两个jvm中是很容易出现线程冲突的;

那么我们可以使用UUID来区分不同的服务，然后用不同的线程id来区分不同的线程 ;

两者结合就一定能够标识线程，不同线程一定不一样；

这里修改一下我们的代码逻辑即可 :

package com.hmdp.utils;

import cn.hutool.core.lang.UUID;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

public class SimpleRedisLock implements ILock{

    private String name ; // 业务名称
    private StringRedisTemplate stringRedisTemplate ;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    private static final String KEY_PREFIX = "lock:" ; // 锁的前缀
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-" ;

    /**
     * 尝试获取锁
     * @param timeoutSec : 锁持有的超时时间 ， 过期后自动释放 ;
     * @return true代表获取锁成功 ， false代表获取锁失败 ;
     */
    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取当前线程id :
        String threadId = ID_PREFIX + Thread.currentThread().getId() ;
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name  , threadId , timeoutSec, TimeUnit.SECONDS) ;
        return Boolean.TRUE.equals(success);
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        // 获取线程标识
        String threadId = ID_PREFIX + Thread.currentThread().getId() ;
        // 获取锁中的指示
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
        // 判断是否一致
        if(threadId.equals(id)){
            // 释放锁
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }

    }
}

6 . 分布式锁的原子性问题

之前的逻辑在极端情况下还是可能出现问题 :

因为判断锁标识和释放锁是两个动作，在这两个动作之之间可能发生阻塞 :

所以我们必须要保证两个动作是原子性的 :

7 . Lua脚本解决多条命令原子性问题

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站：Lua 教程 | 菜鸟教程，这里重点介绍Redis提供的调用函数，我们可以使用lua去操作redis，又能保证他的原子性，这样就可以实现拿锁比锁删锁是一个原子性动作了，作为Java程序员这一块并不作一个简单要求，并不需要大家过于精通，只需要知道他有什么作用即可。

这里重点介绍Redis提供的调用函数，语法如下：

redis.call('命令名称', 'key', '其它参数', ...)

例如，我们要执行set name jack，则脚本是这样：

# 执行 set name jack
redis.call('set', 'name', 'jack')

例如，我们要先执行set name Rose，再执行get name，则脚本如下：

# 先执行 set name jack
redis.call('set', 'name', 'Rose')
# 再执行 get name
local name = redis.call('get', 'name')
# 返回
return name

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

例如我们要执行命令set name jack命令如下 :

如果要先执行set name,再执行get name ;

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

例如，我们要执行 redis.call('set', 'name', 'jack') 这个脚本，语法如下：

如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组获取这些参数：

先写好脚本如下 :

-- 获取锁中的线程标识 get key
local id = redis.call('get' ,KEYS[1])
-- 比较线程标识与锁中标识是否一致
if(id == ARGV[1]) then
    -- 释放锁 del key
    return redis.call('del',KEYS[1])
end
return 0

8 . 利用Java代码调用Lua脚本改造分布式锁

lua脚本本身并不需要大家花费太多时间去研究，只需要知道如何调用，大致是什么意思即可，所以在笔记中并不会详细的去解释这些lua表达式的含义。

我们的RedisTemplate中，可以利用execute方法去执行lua脚本，参数对应关系就如下图：

释放锁只需要调用脚本即可 :

    @Override
    public void unlock() {

        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name) ,
                ID_PREFIX + Thread.currentThread().getId()
        ) ;
    }

小总结：

基于Redis的分布式锁实现思路：

利用set nx ex获取锁，并设置过期时间，保存线程标示
释放锁时先判断线程标示是否与自己一致，一致则删除锁
- 特性：
  - 利用set nx满足互斥性
  - 利用set ex保证故障时锁依然能释放，避免死锁，提高安全性
  - 利用Redis集群保证高可用和高并发特性

笔者总结：我们一路走来，利用添加过期时间，防止死锁问题的发生，但是有了过期时间之后，可能出现误删别人锁的问题，这个问题我们开始是利用删之前通过拿锁，比锁，删锁这个逻辑来解决的，也就是删之前判断一下当前这把锁是否是属于自己的，但是现在还有原子性问题，也就是我们没法保证拿锁比锁删锁是一个原子性的动作，最后通过lua表达式来解决这个问题

但是目前还剩下一个问题锁不住，什么是锁不住呢，你想一想，如果当过期时间到了之后，我们可以给他续期一下，比如续个30s，就好像是网吧上网，网费到了之后，然后说，来，网管，再给我来10块的，是不是后边的问题都不会发生了，那么续期问题怎么解决呢，可以依赖于我们接下来要学习redission啦

测试逻辑：

第一个线程进来，得到了锁，手动删除锁，模拟锁超时了，其他线程会执行lua来抢锁，当第一天线程利用lua删除锁时，lua能保证他不能删除他的锁，第二个线程删除锁时，利用lua同样可以保证不会删除别人的锁，同时还能保证原子性。