高并发情况下加锁-本地锁、分布式锁实现

前言

在高并发情况下，要保证服务端的性能，那么会采用缓存来提高服务端的性能，如百万请求访问一个查询的接口，这个接口做了缓存，但是不能保存并发同时到达接口时缓存中也没有数据，恰巧这百万的并发又进入到数据库，那么这时数据库压力过大，导致数据库崩溃，导致服务的不可用，乃至整个系统的崩溃，那么这是由于并发同时绕过了缓存判断直接进入到数据库导致的，这时就可以针对这个并发问题进行加锁

本地锁

单体项目时可以这么做–伪代码

public R getData {
        /**
         * 将数据库的多次查询变为一次查询
         * SpringBoot 所有的组件在容器中默认都是单例的，使用 synchronized (this) 可以实现加锁
         */
        synchronized (this) {
            /**
             * 得到锁之后 应该再去缓存中确定一次，如果没有的话才需要继续查询
             * 假如有100W个并发请求，首先得到锁的请求开始查询，此时其他的请求将会排队等待锁
             * 等到获得锁的时候再去执行查询，但是此时有可能前一个加锁的请求已经查询成功并且将结果添加到了缓存中
             */

            //TODO 数据库查询数据操作
        }
    }

实操

导入Redis相关依赖

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>io.lettuce</groupId>
                    <artifactId>lettuce-core</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
        </dependency>

不加同步代码块

@RequestMapping("/nativeLock")
    public R list(@RequestParam Map<String, Object> params) throws InterruptedException {
            String data = stringRedisTemplate.opsForValue().get("data");
            if (!StringUtils.isEmpty(data)) {
                log.info("缓存中拿数据");
                return R.ok().setData(data);
            }
            log.error("进入数据库查数据");
            //模拟数据库操作
            Thread.sleep(200);//模拟数据库查询耗时
            List<String> dbData=testService.getDataList();
            String cache = JSON.toJSONString(dbData);
            stringRedisTemplate.opsForValue().set("data", cache, 1, TimeUnit.DAYS);
            return  R.ok().setData(dbData);
    }

使用JMeter模拟并发测试

这里我使用JMeter模拟了1000个并发请求，从控制台打印结果来看，并没有像理论上那样只查询一次数据库，然后将第一个请求查出的数据缓存到redis中，后面的请求都从redis中拿，这是由于有每个并发都对应一个线程，每个线程每走一行代码都要竞争CPU的当部分线程都卡在String data = stringRedisTemplate.opsForValue().get(“data”);这是Redis中还没有缓存数据，那么这些卡在String data = stringRedisTemplate.opsForValue().get(“data”);这一行代码的线程都得到的是空数据，所以这些得到空数据的都会进入Mysql中查询！！！

优化-加同步代码

@RequestMapping("/nativeLock")
    public R list(@RequestParam Map<String, Object> params) throws InterruptedException {
        String data = stringRedisTemplate.opsForValue().get("data");
        if (StringUtils.isEmpty(data)) {
            return getCatalogJsonFromDb();
        }
        log.info("缓存中拿数据");
        return R.ok().setData(data);
    }

    public R getCatalogJsonFromDb() throws InterruptedException {
        synchronized (this){
            得到锁后再次确认缓存中，是否存在数据
            String data = stringRedisTemplate.opsForValue().get("data");
            if (StringUtils.isEmpty(data)) {
                log.error("进入数据库查数据");
                List<String> dbData = testService.getDataList();
                stringRedisTemplate.opsForValue().set("data", String.valueOf(dbData));
                return R.ok().setData(dbData);
           }
           return R.ok().setData(data);
        }
    }

使用JMeter模拟并发测试

这里加了synchronized后确实能锁住每个线程，当并发进入后，第一个请求抢到锁后，其他的线程请求就会等待，直到第一个并发请求查询缓存–>判断缓存–>数据库存入缓存–>释放锁；后面的并发请求才能进来，那么后面的并发请求进来的时候去检查Redis缓存的时候就都有值了，那么就只会从Redis中取值，并不会进入Mysql中查询

注意：
这里实际测试下来，没有加同步代码的吞吐量为900多，二加了同步代码的吞吐量只有300多！！！

分布式环境下的情况

这里我使用gateway来做代理，当然咯也可以使用nginx来做代理

将服务都启动，开始并发测试

13001

13002

13003

13004

13005

缺点

分析上面五个服务情况，本地锁只能锁住当前进程，在分布式架构环境下锁不住所有的服务请求，难免每个服务还是会对数据库进行一次IO

分布式锁

废话不多说，直接上代码分析

1.0版本

@RequestMapping("/distributedLock")
    public R distributedLock() throws InterruptedException {
        String data = stringRedisTemplate.opsForValue().get("data");
        if (StringUtils.isEmpty(data)) {
            R res = a();
            return res;
        }
        log.info("缓存中拿数据");
        return R.ok().setData(data);
    }
     //1.0  
    public R a() throws InterruptedException {
        //1.抢占分布式锁，到redis中占坑
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111");
        if (lock) {
            //加锁成功...执行业务逻辑
            String data = stringRedisTemplate.opsForValue().get("data");//占到坑了在来确认一下缓存中是否存在数据
            if (StringUtils.isEmpty(data)) {
                List<String> dbData = testService.getDataList();
                stringRedisTemplate.opsForValue().set("data", String.valueOf(dbData));
                data= String.valueOf(dbData);
            }
            stringRedisTemplate.delete("lock");//缓存中也需要释放锁，不然其他服务无法得到锁
            return R.ok().setData(data);
        } else {
            Thread.sleep(200);//防止自旋频率过高
            return a();//没占到坑开始等待锁自旋
        }
    }
    
    //模拟数据库
    @Override
    public List<String> getDataList() {
        try {
            System.err.println("进入数据库查数据");
            Thread.sleep(200);//模拟数据库查询耗时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return l;
    }

1.0存在问题

2.0-版本

//2.0
    public R b() throws InterruptedException {
        //1.抢占分布式锁，到redis中占坑
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111");
        if (lock) {
            //加锁成功...执行业务逻辑
            stringRedisTemplate.expire("lock",30,TimeUnit.SECONDS);//设置过期时间-避免死锁
            String data = stringRedisTemplate.opsForValue().get("data");//占到坑了在来确认一下缓存中是否存在数据
            if (StringUtils.isEmpty(data)) {
                List<String> dbData = testService.getDataList();
                stringRedisTemplate.opsForValue().set("data", String.valueOf(dbData));
                data= String.valueOf(dbData);
            }
            stringRedisTemplate.delete("lock");//缓存中也需要释放锁，不然其他服务无法得到锁
            return R.ok().setData(data);
        } else {
            Thread.sleep(200);//模拟数据库查询耗时
            return b();//没占到坑开始等待锁自旋
        }
    }

2.0存在问题

3.0版本

//3.0
    public R c() throws InterruptedException {
        //1.抢占分布式锁，到redis中占坑
        //避免死锁-设置过期时间-必须和加锁是同步的，原子的
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111",30,TimeUnit.SECONDS);
        if (lock) {
            //加锁成功...执行业务逻辑
            String data = stringRedisTemplate.opsForValue().get("data");//占到坑了在来确认一下缓存中是否存在数据
            if (StringUtils.isEmpty(data)) {
                List<String> dbData = testService.getDataList();
                stringRedisTemplate.opsForValue().set("data", String.valueOf(dbData));
                data= String.valueOf(dbData);
            }
            stringRedisTemplate.delete("lock");//缓存中也需要释放锁，不然其他服务无法得到锁
            return R.ok().setData(data);
        } else {
            Thread.sleep(200);//模拟数据库查询耗时
            return b();//没占到坑开始等待锁自旋
        }
    }

3.0版本存在问题

4.0版本

//4.0
    public R d() throws InterruptedException {
        //1.抢占分布式锁，到redis中占坑
        //避免死锁-设置过期时间-必须和加锁是同步的，原子的
        String uuid = UUID.randomUUID().toString();//给锁加上唯一标识，避免业务耗时过长，将其他线程占用的锁删除
        //当第一个线程抢到锁后开始执行逻辑代码，既定这个逻辑代码执行需要40秒，那么这里lock值设置了30秒，
        // 当到了30秒的时候，这个lock锁将自动过期，同时也会释放lock的锁，那么在这里等待的其他线程就会进入，
        // 此时将会有多个线程在执行数据库操作，那么这里就用uuid来解决误删其他线程的锁的问题
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid, 30, TimeUnit.SECONDS);
        if (lock) {
            //加锁成功...执行业务逻辑
            String data = stringRedisTemplate.opsForValue().get("data");//占到坑了在来确认一下缓存中是否存在数据
            if (StringUtils.isEmpty(data)) {
                List<String> dbData = testService.getDataList();
                stringRedisTemplate.opsForValue().set("data", String.valueOf(dbData));
                data= String.valueOf(dbData);
            }
            String lockFromRedis = stringRedisTemplate.opsForValue().get("lock");
            if (uuid.equals(lockFromRedis))//只删除自己的锁
                stringRedisTemplate.delete("lock"); // 删除锁
            return R.ok().setData(data);
        } else {
            Thread.sleep(200);//模拟数据库查询耗时
            return b();//没占到坑开始等待锁自旋
        }
    }

4.0版本存在问题

5.0版本

//5.0
    public R e() throws InterruptedException {
        //1.抢占分布式锁，到redis中占坑
        //避免死锁-设置过期时间-必须和加锁是同步的，原子的
        String uuid = UUID.randomUUID().toString();//给锁加上唯一标识，避免业务耗时过长，将其他线程占用的锁删除
        //当第一个线程抢到锁后开始执行逻辑代码，既定这个逻辑代码执行需要40秒，那么这里lock值设置了30秒，
        // 当到了30秒的时候，这个lock锁将自动过期，同时也会释放lock的锁，那么在这里等待的其他线程就会进入，
        // 此时将会有多个线程在执行数据库操作，那么这里就用uuid来解决误删其他线程的锁的问题
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid, 30, TimeUnit.SECONDS);
        if (lock) {
            //加锁成功...执行业务逻辑
            String data = stringRedisTemplate.opsForValue().get("data");//占到坑了在来确认一下缓存中是否存在数据
            if (StringUtils.isEmpty(data)) {
                List<String> dbData = testService.getDataList();
                stringRedisTemplate.opsForValue().set("data", String.valueOf(dbData));
                data= String.valueOf(dbData);
            }
            //在获取lock的值+比对成功删除的时候必须是原子性操作，这里需要使用lua脚本解锁 http://redis.cn/commands/set.html 脚本官网
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Collections.singletonList("lock"), uuid);
            //String lockFromRedis = stringRedisTemplate.opsForValue().get("lock");
            //if (uuid.equals(lockFromRedis))//只删除自己的锁
            //    stringRedisTemplate.delete("lock"); // 删除锁
            return R.ok().setData(data);
        } else {
            Thread.sleep(200);//模拟数据库查询耗时
            return b();//没占到坑开始等待锁自旋
        }
    }

最终版

这里只有一个问题，就是锁的时间自动续期问题，这里我们可以把过期时间设置的长一些

//最终版本
    public R f() throws InterruptedException {
        //1.抢占分布式锁，到redis中占坑
        //避免死锁-设置过期时间-必须和加锁是同步的，原子的
        String uuid = UUID.randomUUID().toString();//给锁加上唯一标识，避免业务耗时过长，将其他线程占用的锁删除
        //当第一个线程抢到锁后开始执行逻辑代码，既定这个逻辑代码执行需要40秒，那么这里lock值设置了30秒，
        // 当到了30秒的时候，这个lock锁将自动过期，同时也会释放lock的锁，那么在这里等待的其他线程就会进入，
        // 此时将会有多个线程在执行数据库操作，那么这里就用uuid来解决误删其他线程的锁的问题
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid, 30, TimeUnit.SECONDS);
        if (lock) {
            //加锁成功...执行业务逻辑
            String data = stringRedisTemplate.opsForValue().get("data");//占到坑了在来确认一下缓存中是否存在数据
            try {
                if (StringUtils.isEmpty(data)) {
                    List<String> dbData = testService.getDataList();
                    stringRedisTemplate.opsForValue().set("data", String.valueOf(dbData));
                    //在获取lock的值+比对成功删除的时候必须是原子性操作，这里需要使用lua脚本解锁 http://redis.cn/commands/set.html 脚本官网
                    data= String.valueOf(dbData);
                }
            }finally {
                //在获取lock的值+比对成功删除的时候必须是原子性操作，这里需要使用lua脚本解锁 http://redis.cn/commands/set.html 脚本官网
                String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Collections.singletonList("lock"), uuid);
            }
            return R.ok().setData(data);
        } else {
            Thread.sleep(200);//模拟数据库查询耗时
            return b();//没占到坑开始等待锁自旋
        }
    }

查看最终版的运行

13001

13002

13003

搞定！！！

缺点

分布式锁性能相比本地锁要差一点，流程也麻烦点，属于重量级锁

核心

redis实现分布式锁的核心点就在于加锁时设置值，过期时间必须是原子性，删除锁的时候查询锁，删除锁也必须hi原子性的！！！