利用redis实现分布式锁的一种思路

来源:http://www.chinese-glasses.com 作者:Web前端 人气:87 发布时间:2020-03-31
摘要:时间: 2019-10-19阅读: 110标签: 面试 在一次工作中,出现了一个小意外,一个存在redis写与读的方法中得出的结果总是和预想的不一样,后面仔细思考了以后,发现是因为集群中各个节点都

时间: 2019-10-19阅读: 110标签: 面试

在一次工作中,出现了一个小意外,一个存在redis写与读的方法中得出的结果总是和预想的不一样,后面仔细思考了以后,发现是因为集群中各个节点都使用共享的缓存、队列这些,有些场景中各个节点之间可能会发生资源竞争,可能会发各个节点之间的“线程不安全问题”,单机中,可以使用锁来解决,在分布式环境下,就要用到分布式锁了,当时网上查了查,发现了好多大神给出的思路和代码,自己试着实现了一下,在这把思路分享给大家。

WEB服务,尤其是分布式的WEB服务,通常会面临需要实现安全高效的全局锁的服务场景。而redis的诸多特性,使得借助其实现的全局锁显得如此简单。本篇文章借助小明应聘魔笛手研发岗位的面试,层层递进的来介绍一下如何通过redis实现全局锁。

Redis可以做分布式锁的条件

因为redis是单线程的,并行访问的指令在他内部会是串行执行的,所以redis可以作为实现分布式锁的技术。

面试官老张:看你简历上说,之前的项目中使用了redis,你知道redis为什么是单线程的吗?

所用的命令

在前几篇文章中,已经介绍了Redis的基本类型与其相应的操作,有不太了解的朋友可以去《Redis常用类型知多少?》去看看。在实现分布式锁中,主要用到了redis的字符串类型数据结构,以及以下的操作:

小明:多进程或者多线程更多的为了高效利用CPU,但是却对开发人员提出了更高的要求。而redis作为内存型数据库,其性能瓶颈往往不是CPU,而是内存和网络。单线程对于redis来说已经足够快了,也就没有必要再引入复杂的多任务模型。

set key value [10bet,ex 秒数] [px 毫秒数] [nx/xx] : 保存key-value

ex、px指的是key的有效期。
nx:表示当key不存在时,创建key并保存value
xx:表示当key存在时,保存value

localhost:0>set hello world
OK
localhost:0>set hello world ex 100
OK
localhost:0>set hello world px 1000
OK
这里主要看看 nx与xx的效果
localhost:0>set hello worldnx nx 
NULL
localhost:0>set hello worldxx xx 
OK
因为hello这个key已经存在,所以带上参数nx后,返回了NULL,没有操作成功。
带上xx这个参数后,现在hello这个key的value已经被替换成了worldxx

实现中用到了set指令的nx参数,作用是保存一个key-value,如果key不存在时,创建key并保存value,如果key存在是,不做操作返回NULL,在java代码中,这个命令被封装成了Jedis的一个方法,我自己在代码中又做了一个封装:

    /**
     * 如果key不存在则建立返回1
     * 如果key存则不作操作,返回0
     * @param key
     * @param value
     * @return
     */
    public Long setNx(final String key,final String value){
        return this.execute(new Function<Long, Jedis>() {
            @Override
            public Long callback(Jedis jedis) {
                //set命令带nx参数在Jedis中被封装成了setnx方法
                return jedis.setnx(key, value);
            }
        });
    }

主要用来获取锁,setnx返回1,证明获取到了锁,返回0证明锁已经存在。

注:官方文档有提到单线程的redis就可以达到10W的QPS。

expire key 整数值:设置key的生命周期以秒为单位

主要用来设置锁的过期时间,贴出java中封装的方法expire:

   /**
     * 设置key的过期时间
     * @param key
     * @param exp
     * @return
     */
    public Long expire(final String key,final int exp){
        return this.execute(new Function<Long, Jedis>() {
            @Override
            public Long callback(Jedis jedis) {
                return jedis.expire(key, exp);
            }
        });
    }

面试官老张:不错,我再问一个简单的问题,通过什么命令为一个key指定字符串的值。

get key : 取出key对应的value

取出存进去的锁值,贴出java中封装的方法get:

   /**
     * 获取String类型的值
     *
     * @param key
     * @return
     */
    public String get(final String key) {
        return this.execute(new Function<String, Jedis>() {
            @Override
            public String callback(Jedis jedis) {
                return jedis.get(key);
            }
        });
    }

小明:set命令。

getset key newvalue:返回key的旧value,把新值newvalue存进去
localhost:0>getset hello wolrd
wogldxx
localhost:0>get hello
wolrd

用于获取到锁以后把新的锁值存进去,贴出java中封装的方法getset:

   /**
     * 获取并返回旧值,在设置新值
     * @param key
     * @param value
     * @return
     */
    public String getSet(final String key, final String value){
        return this.execute(new Function<String, Jedis>() {
            @Override
            public String callback(Jedis jedis) {
                return jedis.getSet(key, value);
            }
        });
    }

面试官老张:恩,这个问题有点太简单了哈,那你能介绍一下set命令都可以接收什么参数吗?

实现的思路

1.先利用当前的时间戳与锁的过期时间相加后得出锁将要过期的时间,并把这个时间作为value,锁作为key调用setnx方法,如果返回1,证明原来这个key,也就是锁不存在,获取锁成功,则调用expire方法设置锁的超时时间,返回获取锁成功。
2.如果setnx返回0,证明原来锁存在,没有获取到锁,然后阻塞等锁的获取。(我原来自己的思路就考虑到第2步就结束了,后面看了大神的实现思路,发现我的想法欠妥,下面说一下大神的第3步)。
3.如果setnx返回0,证明原来锁存在,没有获取到锁,然后调用get方法,获取第1步存进去的value,就是锁将要过期的时间,与当前的时间戳比较,如果如果当前的时间戳大于锁将要过期的时间,证明锁已经过期,调用getset方法,把新的时间存进去,返回获取锁成功。这样做主要是应对expire执行失败,或者服务器重启的情况下出现的锁无法释放的情况。
下面把按照大神思路实现的代码贴出来,实现了阻塞机制的锁,也实现了一个排它锁:

import com.eduapi.common.component.RedisComponent;
import com.eduapi.common.util.BeanUtils;
import org.apache.commons.lang3.StringUtils;

/**
 * @Description: 利用redis实现分布式锁.
 * @Author: ZhaoWeiNan .
 * @CreatedTime: 2017/3/20 .
 * @Version: 1.0 .
 */
public class RedisLock {

    private RedisComponent redisComponent;

    private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;

    private String lockKey;

    /**
     * 锁超时时间,防止线程在入锁以后,无限的执行等待
     */
    private int expireMillisCond = 60 * 1000;

    /**
     * 锁等待时间,防止线程饥饿
     */
    private int timeoutMillisCond = 10 * 1000;

    private volatile boolean isLocked = false;

    public RedisLock(RedisComponent redisComponent, String lockKey) {
        this.redisComponent = redisComponent;
        this.lockKey = lockKey;
    }

    public RedisLock(RedisComponent redisComponent, String lockKey, int timeoutMillisCond) {
        this(redisComponent, lockKey);
        this.timeoutMillisCond = timeoutMillisCond;
    }

    public RedisLock(RedisComponent redisComponent, String lockKey, int timeoutMillisCond, int expireMillisCond) {
        this(redisComponent, lockKey, timeoutMillisCond);
        this.expireMillisCond = expireMillisCond;
    }

    public RedisLock(RedisComponent redisComponent, int expireMillisCond, String lockKey) {
        this(redisComponent, lockKey);
        this.expireMillisCond = expireMillisCond;
    }

    public String getLockKey() {
        return lockKey;
    }

    /**
     * 获得 lock. (把大神的思路粘过来了)
     * 实现思路: 主要是使用了redis 的setnx命令,缓存了锁.
     * reids缓存的key是锁的key,所有的共享, value是锁的到期时间(注意:这里把过期时间放在value了,没有时间上设置其超时时间)
     * 执行过程:
     * 1.通过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功获得锁
     * 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
     *
     * @return true if lock is acquired, false acquire timeouted
     * @throws InterruptedException in case of thread interruption
     */
    public synchronized boolean lock() throws InterruptedException {
        int timeout = timeoutMillisCond;

        boolean flag = false;

        while (timeout > 0){
            //设置所得到期时间
            Long expires = System.currentTimeMillis() + expireMillisCond;
            String expiresStr = BeanUtils.convertObject2String(expires);

            //原来redis里面没有锁,获取锁成功
            if (this.redisComponent.setNx(lockKey,expiresStr) > 0){
                //设置锁的过期时间
                this.redisComponent.expire(lockKey,expireMillisCond);
                isLocked = true;
                return true;
            }

            flag = compareLock(expiresStr);

            if (flag){
                return flag;
            }

            timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;

            /*
                延迟100 毫秒,  这里使用随机时间可能会好一点,可以防止饥饿进程的出现,即,当同时到达多个进程,
                只会有一个进程获得锁,其他的都用同样的频率进行尝试,后面有来了一些进程,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
                使用随机的等待时间可以一定程度上保证公平性
             */
            Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
        }
        return false;
    }

    /**
     * 排他锁。作用相当于 synchronized 同步快
     * @return
     * @throws InterruptedException
     */
    public synchronized boolean excludeLock() {

        //设置所得到期时间
        long expires = System.currentTimeMillis() + expireMillisCond;
        String expiresStr = BeanUtils.convertObject2String(expires);

        //原来redis里面没有锁,获取锁成功
        if (this.redisComponent.setNx(lockKey,expiresStr) > 0){
            //设置锁的过期时间
            this.redisComponent.expire(lockKey,expireMillisCond);
            isLocked = true;
            return true;
        }

        return compareLock(expiresStr);
    }

    /**
     * 比较是否可以获取锁
     * 锁超时时 获取
     * @param expiresStr
     * @return
     */
    private boolean compareLock(String expiresStr){
        //假如两个线程走到这里
        //因为redis是单线程的获取到
        // A线程获取  currentValueStr = 1 B线程获取 currentValueStr = 1
        String currentValueStr = this.redisComponent.get(lockKey);

        //锁
        if (StringUtils.isNotEmpty(currentValueStr) && Long.parseLong(currentValueStr) < System.currentTimeMillis()){

            //获取上一个锁到期时间,并设置现在的锁到期时间,
            //只有一个线程才能获取上一个线程的设置时间,因为jedis.getSet是同步的
            //只有A线程 把 2 存进去了。 取出了 1, 对比获得了锁
            //B线程 吧 2存进去了。 获取 2.。对比 没有获得锁,
            String oldValue = this.redisComponent.getSet(lockKey,expiresStr);

            if (StringUtils.isNotEmpty(oldValue) && StringUtils.equals(oldValue,currentValueStr)){

                //防止误删(覆盖,因为key是相同的)了他人的锁——这里达不到效果,这里值会被覆盖,但是因为什么相差了很少的时间,所以可以接受
                //[分布式的情况下]:如过这个时候,多个线程恰好都到了这里,但是只有一个线程的设置值和当前值相同,他才有权利获取锁
                isLocked = true;
                return true;
            }
        }
        return false;
    }

    /**
     * 释放锁
     */
    public synchronized void unlock(){
        if (isLocked){
            this.redisComponent.delete(lockKey);
            isLocked = false;
        }
    }

}

代码中,把大神写的思路站进去了,大神讲的还是很明白的,看看调用的代码:

        //创建锁对象, redisComponent 为redis组件的对象   过期时间  锁的key
        RedisLock redisLock = new RedisLock(redisComponent,1000 * 60,RedisCacheKey.REDIS_LOCK_KEY + now_mm);
        //获取锁
        if (redisLock.excludeLock()){
            try {
                //拿到了锁,读取定时短信有序集合
                set = this.redisComponent.zRangeByScore(RedisCacheKey.MSG_TIME_LIST,0,end);

                if (set != null && set.size() > 0){
                    flag = true;
                }
            } catch (Exception e) {
                LOGGER.error("获取定时短信有序集合异常,异常为{}",e.toString());
            }
...

利用redis实现分布式锁的一种思路就为大家介绍到这里,欢迎大家来交流,指出文中一些说错的地方,让我加深认识。
谢谢大家!

小明:这个我有查阅过,redis的set命令除了指定key和value,还有几个可选参数,分别是:

ex——设置key多少秒之后过期

px——设置key多少毫秒之后过期

本文由10bet发布于Web前端,转载请注明出处:利用redis实现分布式锁的一种思路

关键词:

频道精选

最火资讯