阅读 1362

Redis修行 — 分布式锁

常见的实现方式

  • 基于数据库的分布式锁
  • 基于缓存的分布式锁(redis,memcached等)
  • 基于ZooKeeper的分布式锁(临时有序节点)

本文主要介绍通过Redis自己去实现分布式锁以及使用开源框架Redisson去实现分布式锁,基于数据库和Zookeeper方式简要带过。

特性

  • 互斥性:只能有一个客户端持有锁
  • 防死锁:客户端在持有锁期间崩溃,未能解锁,也有其他方式去解锁,不影响其他客户端获取锁
  • 只有加锁的人才能释放锁

原理

分布式锁本质上可以理解为是一个所有客户端共享的全局变量,当这个全局变量存在时,说明已经有客户端获取到了锁,其他客户端只能等它释放锁(删除这个全局变量)后才能获取到锁(设置全局变量)。

基于Redis实现分布式锁

按照上面的特性和理论,我们整理一下基本思路:

  • 指定一个key作为锁标记,存入Redis中,指定一个唯一的用户标识作为value
  • 当key不存在时才能设置值,确保同一时间只有一个客户端获得锁,满足互斥性特性
  • 设置一个过期时间,防止因系统异常导致没能删除这个key,满足防死锁特性
  • 当处理完业务之后需要清除这个key来释放锁。
  • 清除key时需要校验value值,需要满足只有加锁的人才能释放锁

获取锁

使用以下指令:

SET mylock userId NX PX 10000
复制代码
  • mylock为锁对应的key
  • userId为唯一的用户标识,用于删除时校验
  • NX表示只有当key不存在时才能set成功,确保只有一个客户端能够请求成功
  • PX 10000表示这个锁有一个10秒的自动过期时间

释放锁

当业务完成后删除key来释放锁,可以执行以下lua脚本:

if redis.call("get",KEYS[1]) == ARGV[1then
    return redis.call("del",KEYS[1])
else
    return 0
end
复制代码

执行以上脚本时,需要将mylock作为KEYS[1]传进去,将userId作为ARGV[1]传进去

注意点

  • 必须要给锁加一个过期时间:这样即使中间系统异常了,等过期时间到了,也可以自动释放锁,防止出现死锁现象
  • 获取锁时不能分成先设置key,再设置过期时间两步去执行,错误示例如下:
    # 当key不存在时设置值
    setnx mylock userId
    # 设置过期时间
    expire mylock 10
复制代码

这样会存在一个问题,如果系统在执行完setnx之后异常了,expire指令就无法执行,同样会出现死锁现象

  • 有必要将value设置为一个唯一的用户标识,用于保证所要释放的锁是自己建立的,因为在极端的情况下会出现下列情况:

A成功获取了锁

A在某个操作上被阻塞了很久

A的锁到达过期时间

B获取了锁

A从阻塞中恢复了,执行释放锁操作,把B的锁释放了,导致B操作不受保护

  • 释放锁操作需要保证操作时原子性的,需要通过Lua脚本来实现。它将GET、判断是否相同、DEL三个步骤以一个原子性的方式去完成。如果按逻辑分开执行同样会出现类似上面的问题:

A先判断当前锁的值,确定了是自己建的锁,准备释放锁了

因为网路问题或者系统卡顿导致A被阻塞了

A的锁过期了

B获取锁

A从阻塞中恢复了

A调用DEL释放了B的锁

缺陷

从上面的描述可以看出来,当出现系统阻塞或者网络延迟等情况下,可能业务还没有执行完成,锁就过期自动释放了,这时它的业务操作时不受保护的。

代码实现

本文样例基于SpringBoot实现

pom.xml
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<!-- redis Lettuce 模式 连接池 -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>
复制代码
yml配置文件
spring:
  redis:
    # Redis数据库索引(默认为0)
    database: 0
    # Redis服务器地址
    host: localhost
    # Redis服务器连接端口
    port: 6379
    # Redis服务器连接密码(默认为空)
    # password: admin
    # 连接超时时间(毫秒)
    timeout: 3000ms
    lettuce:
      pool:
        # 连接池最大连接数(使用负值表示没有限制)
        max-active: 20
        # 连接池最大阻塞等待时间(使用负值表示没有限制)
        max-wait: 3000ms
        # 连接池中的最大空闲连接(负数没有限制)
        max-idle: 8
        # 连接池中的最小空闲连接
        min-idle: 0
复制代码
锁操作
@Component
public class RedisLock {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 加锁
     */

    public boolean tryLock(String key, String value) {
        Boolean isLocked = stringRedisTemplate.opsForValue().setIfAbsent(key, value, 5, TimeUnit.SECONDS);
        if (isLocked == null) {
            return false;
        }
        return isLocked;
    }

    /**
     * 解锁
     */

    public Boolean unLock(String key, String value) {
        // 执行 lua 脚本
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        // 指定 lua 脚本
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("redis/unLock.lua")));
        // 指定返回类型
        redisScript.setResultType(Long.class);
        // 参数一:redisScript,参数二:key列表,参数三:arg(可多个)
        Long result = stringRedisTemplate.execute(redisScript, Collections.singletonList(key), value);
        return result != null && result > 0;
    }
}
复制代码

释放锁需要执行Lua脚本,路径为:resources/redis/unLock.lua

if redis.call("get",KEYS[1]) == ARGV[1then
  return redis.call("del",KEYS[1])
else
  return 0
end
复制代码
测试

模拟一个减库存的操作,先在redis中设置库存量50,key为productKey,创建访问接口:

@RestController
@RequestMapping("/redis")
public class RedisController {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private static final String PRODUCT_KEY = "productKey";

    private static final String LOCK_KEY = "redisLock";

    @Autowired
    private RedisLock redisLock;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @GetMapping("/lock")
    public void lockTest() throws InterruptedException {
        // 用户唯一标识
        String lockValue = UUID.randomUUID().toString().replace("-""");
        Random random = new Random();
        int sleepTime;
        while (true) {
            if (redisLock.tryLock(LOCK_KEY, lockValue)) {
                logger.info("[{}]成功获取锁", lockValue);
                break;
            }
            sleepTime = random.nextInt(1000);
            Thread.sleep(sleepTime);
            logger.info("[{}]获取锁失败,{}毫秒后重新尝试获取锁", lockValue, sleepTime);
        }
        // 剩余库存
        String products = stringRedisTemplate.opsForValue().get(PRODUCT_KEY);
        if (products == null) {
            logger.info("[{}]获取剩余库存失败,释放锁:{} @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@", lockValue, redisLock.unLock(LOCK_KEY, lockValue));
            return;
        }
        int surplus = Integer.parseInt(products);
        if (surplus <= 0) {
            logger.info("[{}]库存不足,释放锁:{} ##########################################", lockValue, redisLock.unLock(LOCK_KEY, lockValue));
            return;
        }

        logger.info("[{}]当前库存[{}],操作:库存-1", lockValue, surplus);
        stringRedisTemplate.opsForValue().decrement(PRODUCT_KEY);
        logger.info("[{}]操作完成,开始释放锁,释放结果:{}", lockValue, redisLock.unLock(LOCK_KEY, lockValue));
    }
}
复制代码

启动项目,使用JMeter进行并发测试,设置1秒60次请求,观察控制台输出和最终redis中库存数量

Redisson 实现

Redisson是【Redis官方推荐】官网推荐分布式锁实现的方案。使用起来也很简单。这里只做简单演示,具体可以看官方文档

Redis son 莫非是redis亲儿子的意思

pom.xml

直接引入redisson-spring-boot-starter,它包含了对spring-boot-starter-webspring-boot-starter-data-redis的依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.12.0</version>
</dependency>
复制代码

创建配置文件

@Configuration
public class RedissonConfig {
    /**
     * 这里只配置单节点的,支持集群、哨兵等方式配置
     * 可以用Config.fromYAML加载yml文件中的配置
     */

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://localhost:6379")
                .setDatabase(0);
        return Redisson.create(config);
    }
}
复制代码

注意这里的address需要以 redis://host:port 的格式

创建测试接口

@RestController
@RequestMapping("/redisson")
public class RedissonController {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private static final String PRODUCT_KEY = "productKey";

    private static final String LOCK_KEY = "redissonLock";

    @Autowired
    private RedissonClient redissonClient;

    @RequestMapping("/lock")
    public void lock() {
        RLock lock = redissonClient.getLock(LOCK_KEY);
        // 设置5秒过期时间
        lock.lock(5, TimeUnit.SECONDS);
        String lockValue = lock.toString();
        logger.info("[{}]成功获取锁,开始执行业务。。。", lockValue);

        RAtomicLong atomicLong = redissonClient.getAtomicLong(PRODUCT_KEY);
        long surplus = atomicLong.get();
        if (surplus <= 0) {
            lock.unlock();
            logger.info("[{}]库存不足,释放锁 ##########################################", lockValue);
            return;
        }
        logger.info("[{}]当前库存[{}],库存 -1,剩余库存[{}]", lockValue, surplus, atomicLong.decrementAndGet());

        logger.info("[{}]操作完成,释放锁", lockValue);
        lock.unlock();
    }
}
复制代码

启动项目,使用JMeter进行并发测试,同样设置1秒60次请求,观察控制台输出和最终redis中库存数量

基于数据库实现分布式锁

通过唯一索引的方式

# 建立一张记录锁信息的表
lockName -- 锁名称。 加上唯一索引,确保只能有一个客户端获得锁
creater -- 创建人,只有创建者才能解锁
expire -- 过期时间
复制代码
  • 执行前先插入锁数据,lockName做了唯一性约束,如果多个请求同时提交只会有一个请求提交成功。
  • 执行完后删除锁
  • 可以通过定时任务方式去删除已过期的数据,防止死锁

通过乐观锁的形式

  • 在需要操作的表中加一个字段version
  • 操作任务前先查询到当前version的值
    select version from product where product_name = '电脑'
复制代码
  • 更新数据时,将前面查出来的version的值作为条件
    update product set product_count = product_count - 1version = version + 1 where product_name = '电脑' and version = ${version}
复制代码

这样如果在这期间数据被修改了,那么version的值就不一致了,更新操作会失败。这样就确保了在你业务期间没有其他人修改过数据。

基于 ZooKeeper 的分布式锁

ZooKeeper的分布式锁主要是通过创建临时有序节点的方式实现的:

  • 发起加锁请求,在ZooKeeper中创建一个临时有序节点
  • 判断自己创建的节点是否是最小序号
  • 如果是最小的,则成功获取锁
  • 如果不是最小的,则在它的上一节点加上一个监听器
  • 处理完业务后,释放锁,即删除对应的节点
  • ZooKeeper通知监听这个节点的监听器,你的前面已经没有其他节点了,你可以获取锁了
  • 对应节点获取锁

可以发现,ZooKeeper的方式获取锁是有序的,先请求的先获取锁,而通过redis的方式是无序的,谁先抢到谁获得锁

访问源码

所有代码均上传至Github上,方便大家访问

>>>>>> Redis实现分布式锁 <<<<<<

日常求赞

创作不易,如果各位觉得有帮助,求点赞 支持

求关注


关注下面的标签,发现更多相似文章
评论