redis分布式锁的问题和解决

分布式锁

在分布式环境中，为了保证业务数据的正常访问，防止出现重复请求的问题，会使用分布式锁来阻拦后续请求。具体伪代码如下：

    public void doSomething(String userId){
        User user=getUser(userId);
        if(user==null){
            user.setUserName("xxxxx");
            user.setUserId(userId);
            insert(user);
            return;
        }
        update(user);
    }

```     

上面的代码很简单，查询db中有没有对应的user数据，如果有的话，执行更新操作，如果没有则插入。    

我们知道，上面的代码是线程不安全的，在多线程的环境中，就会出现问题。为了能够保证数据的正确性，在单机环境下，我们可以使用`synchronized`的方法，来保证线程安全，具体修改：  


``` Java   
    public synchronized void doSomething(String userId){
        User user=getUser(userId);
        if(user==null){
            user.setUserName("xxxxx");
            user.setUserId(userId);
            insert(user);
            return;
        }
        update(user);
    }

```     

在单机器的环境下，能够解决线程安全的问题，那在分布式环境下呢？ 这个时候需要用到`分布式锁`.   


分布式锁需要借助其他组件来实现，常用的有`redis`和`zookeeper`。下面我们就用redis的实现，来说明下问题，分布式锁具体的实现方法如下：  

``` Java 
    public  void doSomething(String userId){
        String lock=RedisUtils.get("xxxx"+userId);
        if(StringUtils.isNotEmpty(lock)){//说明当前userId已经被锁定
            return;
        }
        RedisUtils.set("xxxx"+userId,userId,1000);//锁定10s
        User user=getUser(userId);
        if(user==null){
            insert(user);
            RedisUtils.delete("xxxx"+userId);
            return;
        }
        update(user);
        RedisUtils.delete("xxxx"+userId);
        
    }

上面的代码解决了在分布式环境中的并发的问题。但同样需要考虑一个问题，如果insert操作和update操作异常了，分布式锁不会释放，后续的请求还会被拦截。

所以我们再优化，增加对异常的捕获。

    public  void doSomething(String userId){
        try {
                String lock=RedisUtils.get("xxxx"+userId);
                if(StringUtils.isNotEmpty(lock)){//说明当前userId已经被锁定
                    return;
                }
                RedisUtils.set("xxxx"+userId,userId,1000);//锁定1s
                User user=getUser(userId);
                if(user==null){
                    insert(user);
                    return;
                }
                update(user);
        }
        catch(Exception ex){

        }
        finally{
            RedisUtils.delete("xxxx"+userId);
        }
    }

现在即使是程序异常了，锁会自动释放。但redis的get和set也会存在并发问题，我们再继续优化，使用redis中的setnx方法。

    public  void doSomething(String userId){
        try {
                boolean lock=RedisUtils.setnx("xxxx"+userId,userId,1000);//锁定1s
                if(!lock){//说明当前userId已经被锁定
                    return;
                }
                User user=getUser(userId);
                if(user==null){
                    insert(user);
                    return;
                }
                update(user);
        }
        catch(Exception ex){

        }
        finally{
            RedisUtils.delete("xxxx"+userId);
        }
    }

上面的代码好像没有什么问题了，但也存在很大的隐患。
我们分析下，假设第一个请求过来，执行锁定成功，程序开始运行，但是insert和update操作阻塞了1s，第二个请求过来，锁的缓存已经过期，第二个执行锁定成功，这个时候第一个请求完成了锁被释放，第二个请求的锁就被第一次请求释放了，第三次的请求就会造成线程不安全问题。

怎么再去优化呢？问题主要是出现在第一次请求误删锁的问题，所以我们在移除锁的时候要判断能否移除。

思路：我们在锁定的时候，value使用当前的时间戳，删除时判断是否过期如果不过期就不要删除，具体代码如下：

    public  void doSomething(String userId){
        try {
                boolean lock=RedisUtils.setnx("xxxx"+userId,LocalDateTime.now(),1000);//锁定10s
                if(!lock){//说明当前userId已经被锁定
                    return;
                }
                User user=getUser(userId);
                if(user==null){
                    insert(user);
                    return;
                }
                update(user);
        }
        catch(Exception ex){

        }
        finally{
            LocalDateTime lockTIme=    RedisUtils.get("xxxx"+userId);
            if(lockTIme.compare(LocalDateTime.now())<0){
                //说明已经过期，可以删除key
                RedisUtils.delete("xxxx"+userId);
            }
        }
    }

这样即使出现阻塞，第二次的时间戳覆盖了第一次的锁定，这样即使第一次完成了，也不会释放锁。