分布式锁

问题描述

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！

分布式锁主流的实现方案：

基于数据库实现分布式锁
基于缓存（Redis等）
基于Zookeeper

每一种分布式锁解决方案都有各自的优缺点：

性能：redis最高
可靠性：zookeeper最高

这里，我们就基于redis实现分布式锁。

解决方案：使用redis实现分布式锁

利用setnx命令上锁

shell

setnx user 20

缺点：上锁后一直不释放锁，就会一直等待。缺点：上锁后服务器挂掉了，没来的及设置过期时间。

利用set user 10 nx ex 12命令上锁

shell

#nx 表示锁，ex 表示过期时间
set user 10 nx ex 12

总结

补充

redis:命令

shell

 set sku:1:info “OK” NX PX 10000

EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。 PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。 NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。 XX ：只在键已经存在时，才对键进行设置操作。

多个客户端同时获取锁（setnx）
获取成功，执行业务逻辑{从db获取数据，放入缓存}，执行完成释放锁（del）
其他客户端等待重试

代码实现

java

package com.angoubiubiu.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class RedisController {

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/redis")
    public String testRedis() {
        //设置值到redis
        redisTemplate.opsForValue().set("name","lucy");
        //从redis获取值
        String name = (String)redisTemplate.opsForValue().get("name");
        return name;
    }

    /**
     * 测试加锁情况
     */
    @GetMapping("testLock")
    public void testLock(){
        //1获取锁，setne
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111");
        //2获取锁成功、查询num的值
        if(lock){
            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);
            //2.4释放锁，del
            redisTemplate.delete("lock");

        }else{
            //3获取锁失败、每隔0.1秒再获取
            try {
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 测试不加锁情况
     */
    @GetMapping("noLock")
    public void noLock(){

            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);


    }

}

测试加锁

Redis: set num 0 利用ab测试工具模拟并发

shell

ab -n 100 -c 10 http://192.168.10.1:9001/testLock

此处报 Benchmarking 192.168.10.1 (be patient).....done 建议配redis连接池

shell

# 最大可用连接数（默认为8，负数表示无限）
spring.redis.jedis.pool.max-active=-1
# 从连接池中获取连接最大等待时间（默认为-1，单位为毫秒，负数表示无限）
spring.redis.jedis.pool.max-wait=-1
# 最小空闲连接数（默认为0，该值只有为正数才有用）
spring.redis.jedis.pool.min-idle=0
# 最大空闲连接数（默认为8，负数表示无限）
spring.redis.jedis.pool.max-idle=-1

结果：100个请求 num的个数就有100个成功

测试不加锁

ab -n 100 -c 10 [http://192.168.10.1:9001/noLock](http://192.168.10.1:9001/noLock)

结果：100个请求 num的个数只有21个

上面分布式锁已基本实现，但还是有些问题

问题：setnx刚好获取到锁，业务逻辑出现异常，导致锁无法释放
解决：设置过期时间，自动释放锁。

优化之设置锁的过期时间

java

package com.angoubiubiu.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

@RestController
public class RedisController {

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/redis")
    public String testRedis() {
        //设置值到redis
        redisTemplate.opsForValue().set("name","lucy");
        //从redis获取值
        String name = (String)redisTemplate.opsForValue().get("name");
        return name;
    }

    /**
     * 测试加锁情况
     */
    @GetMapping("testLock")
    public void testLock(){
        //1获取锁，setne
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111",3, TimeUnit.SECONDS);//3秒之后过期
        //2获取锁成功、查询num的值
        if(lock){
            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);
            //2.4释放锁，del
            redisTemplate.delete("lock");

        }else{
            //3获取锁失败、每隔0.1秒再获取
            try {
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 测试不加锁情况
     */
    @GetMapping("noLock")
    public void noLock(){

            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);


    }

}

压力测试肯定也没有问题。自行测试

问题：可能会释放其他服务器的锁。

场景：如果业务逻辑的执行时间是7s。执行流程如下

index1业务逻辑没执行完，3秒后锁被自动释放。
index2获取到锁，执行业务逻辑，3秒后锁被自动释放。
index3获取到锁，执行业务逻辑
index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。最终等于没锁的情况。
解释：当a服务器先上锁之后，执行相关操作，此时a服务器卡顿（这里就不会手动释放锁)，此时锁过期后，b服务器拿到锁，执行具体操作，然后a服务器卡顿结束，手动释放锁，这里就会把b服务器的锁给释放掉

解决方案：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁

优化之UUID防误删

java

package com.angoubiubiu.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
public class RedisController {

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/redis")
    public String testRedis() {
        //设置值到redis
        redisTemplate.opsForValue().set("name","lucy");
        //从redis获取值
        String name = (String)redisTemplate.opsForValue().get("name");
        return name;
    }

    /**
     * 测试加锁情况
     */
    @GetMapping("testLock")
    public void testLock(){
        String uuid= UUID.randomUUID().toString().replace("-","");
        //1获取锁，setne
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3, TimeUnit.SECONDS);//3秒之后过期
        //2获取锁成功、查询num的值
        if(lock){
            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);
            //如果当前uuid 和 锁里面存的uuid相同才给予释放锁
            if(uuid.equals(redisTemplate.opsForValue().get(lock))){
                //2.4释放锁，del
                redisTemplate.delete("lock");
            }

        }else{
            //3获取锁失败、每隔0.1秒再获取
            try {
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 测试不加锁情况
     */
    @GetMapping("noLock")
    public void noLock(){

            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);


    }

}

问题：删除操作缺乏原子性。

场景：

index1执行删除时，查询到的lock值确实和uuid相等

shell

uuid=v1
set(lock,uuid)；

index1执行删除前，lock刚好过期时间已到，被redis自动释放在redis中没有了lock，没有了锁。
index2获取了lock index2线程获取到了cpu的资源，开始执行方法。

shell

uuid=v2
set(lock,uuid)；

index1执行删除，此时会把index2的lock删除 index1 因为已经在方法中了，所以不需要重新上锁。index1有执行的权限。index1已经比较完成了，这个时候，开始执行, 删除的index2的锁！

解释：当a服务器上锁后，进行具体操作，对比uuid之后（对比完后)准备释放锁，此时a服务器的锁过期了，b服务器上锁后进行具体操作，由于a服务器已经对比过uuid了，此时直接释放锁，会把b服务器的锁给释放掉

优化之LUA脚本保证删除的原子性

java

package com.angoubiubiu.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
public class RedisController {

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/redis")
    public String testRedis() {
        //设置值到redis
        redisTemplate.opsForValue().set("name","lucy");
        //从redis获取值
        String name = (String)redisTemplate.opsForValue().get("name");
        return name;
    }

    /**
     * 测试加锁情况
     */
    @GetMapping("testLock")
    public void testLock(){
        String uuid= UUID.randomUUID().toString().replace("-","");
        //1获取锁，setne
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3, TimeUnit.SECONDS);//3秒之后过期
        //2获取锁成功、查询num的值
        if(lock){
            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);
            //如果当前uuid 和 锁里面存的uuid相同才给予释放锁
            if(uuid.equals(redisTemplate.opsForValue().get(lock))){
                //2.4释放锁，del
                redisTemplate.delete("lock");
            }

        }else{
            //3获取锁失败、每隔0.1秒再获取
            try {
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 测试不加锁情况
     */
    @GetMapping("noLock")
    public void noLock(){

            Object value = redisTemplate.opsForValue().get("num");
            //2.1判断num为空return
            if(StringUtils.isEmpty(value)){
                return;
            }
            //2.2有值就转成成int
            int num = Integer.parseInt(value+"");
            //2.3把redis的num加1
            redisTemplate.opsForValue().set("num", ++num);


    }
    @GetMapping("testLockLua")
    public void testLockLua() {
        //1 声明一个uuid ,将做为一个value 放入我们的key所对应的值中
        String uuid = UUID.randomUUID().toString();
        //2 定义一个锁：lua 脚本可以使用同一把锁，来实现删除！
        String skuId = "25"; // 访问skuId 为25号的商品 100008348542
        String locKey = "lock:" + skuId; // 锁住的是每个商品的数据

        // 3 获取锁
        Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS);

        // 第一种： lock 与过期时间中间不写任何的代码。
        // redisTemplate.expire("lock",10, TimeUnit.SECONDS);//设置过期时间
        // 如果true
        if (lock) {
            // 执行的业务逻辑开始
            // 获取缓存中的num 数据
            Object value = redisTemplate.opsForValue().get("num");
            // 如果是空直接返回
            if (StringUtils.isEmpty(value)) {
                return;
            }
            // 不是空 如果说在这出现了异常！ 那么delete 就删除失败！ 也就是说锁永远存在！
            int num = Integer.parseInt(value + "");
            // 使num 每次+1 放入缓存
            redisTemplate.opsForValue().set("num", String.valueOf(++num));
            /*使用lua脚本来锁*/
            // 定义lua 脚本
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            // 使用redis执行lua执行
            DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
            redisScript.setScriptText(script);
            // 设置一下返回值类型 为Long
            // 因为删除判断的时候，返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型，
            // 那么返回字符串与0 会有发生错误。
            redisScript.setResultType(Long.class);
            // 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
            redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
        } else {
            // 其他线程等待
            try {
                // 睡眠
                Thread.sleep(1000);
                // 睡醒了之后，调用方法。
                testLockLua();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

Lua 脚本详解：

项目中正确使用：

定义key，key应该是为每个sku定义的，也就是每个sku有一把锁。

java

String locKey ="lock:"+skuId; _// 锁住的是每个商品的数据_
Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid,3,TimeUnit.SECONDS);

总结

加锁

java

// 1. 从redis中获取锁,set k1 v1 px 20000 nx
String uuid = UUID.randomUUID().toString();
Boolean lock = this.redisTemplate.opsForValue()
      .setIfAbsent("lock", uuid, 2, TimeUnit.SECONDS);

使用lua释放锁

java

// 2. 释放锁 del
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
// 设置lua脚本返回的数据类型
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
// 设置lua脚本返回类型为Long
redisScript.setResultType(Long.class);
redisScript.setScriptText(script);
redisTemplate.execute(redisScript, Arrays.asList("lock"),uuid);

重试

java

Thread.sleep(500);
testLock();

为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

互斥性。在任意时刻，只有一个客户端能持有锁。
不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。
加锁和解锁必须具有原子性。

软件设计原则

创建型模式

索引

存储过程

事物和锁机制

持久化操作

应用问题解决

DockerFile

Other

分布式锁

问题描述

解决方案：使用redis实现分布式锁

利用setnx命令上锁

利用set user 10 nx ex 12命令上锁

总结

补充

代码实现

测试加锁

测试不加锁

优化之设置锁的过期时间

优化之UUID防误删

优化之LUA脚本保证删除的原子性

总结

加锁

使用lua释放锁

重试

分布式锁 ​

问题描述 ​

解决方案：使用redis实现分布式锁 ​

利用setnx命令上锁 ​

利用set user 10 nx ex 12命令上锁 ​

总结 ​

补充 ​

代码实现 ​

测试加锁 ​

测试不加锁 ​

优化之设置锁的过期时间 ​

优化之UUID防误删 ​

优化之LUA脚本保证删除的原子性 ​

总结 ​

加锁 ​

使用lua释放锁 ​

重试 ​

分布式锁

问题描述

解决方案：使用redis实现分布式锁

利用setnx命令上锁

利用set user 10 nx ex 12命令上锁

总结

补充

代码实现

测试加锁

测试不加锁

优化之设置锁的过期时间

优化之UUID防误删

优化之LUA脚本保证删除的原子性

总结

加锁

使用lua释放锁

重试