redis分布式锁的问题和解决

Posted oceanheaven

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了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的方法,来保证线程安全,具体修改:

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

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

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

    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);
            
        
    

 

 

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

以上是关于redis分布式锁的问题和解决的主要内容,如果未能解决你的问题,请参考以下文章

Redis | 黑马点评 + 思维导图分布式锁

基于Redis的简单分布式锁的原理

Redis分布式锁的原理是啥?如何续期?

redis分布式锁常见问题及解决方案

锁的概念及Redis分布式锁的实现(转)

Redis专题:锁的基本概念到Redis分布式锁实现