Redis实现分布式锁(setnx、getset、incr)以及如何处理超时情况

如果你通过网络搜索分布式锁，最多的就是基于redis的了。基于redis的分布式锁得益于redis的单线程执行机制，单线程在执行上就保证了指令的顺序化，所以很大程度上降低了开发人员的思考设计成本。

一、通过setnx实现

1、setnx key value

当且仅当key不存在，将key的值设置为value，并且返回1；若是给定的key已经存在，则setnx不做任何动作，返回0。

public static boolean setnx(final string key, final string value, final long seconds) {
        return getshardedjedisclient().execute(new shardedjedisaction<boolean>() {
            public boolean doaction(shardedjedis shardedjedis) {
                jedis jedis = (jedis) shardedjedis.getshard(key);
                string result = jedis.set(key, value, "nx", "ex", seconds);
                return "ok".equalsignorecase(result);
            }
        });
    }

2、get key

获取key对应的value值，如果不存在该key，返回0。

public string get(final string key) {
        this.checkisinmulti();
        return (string)this.execute(new smartjedis.action<string>() {
            public string doaction(jedis jedis) {
                return jedis.get(key);
            }
        }, smartjedis.rw.r, key);
    }

3、getset key value

获取key的旧值，将新value放入

public static string getset(final string key, final string value) {
        return getshardedjedisclient().execute(new shardedjedisaction<string>() {
            @override
            public string doaction(shardedjedis shardedjedis) {
                return shardedjedis.getset(key, value);
            }
        });
    }

至此，我们先举个手机三要素验证的列子：(a渠道系统，业务b系统，外部厂商c系统)
(1)b业务系统调用a渠道系统，验证传入的手机、身份证、号码三要素是否一一致。
(2)a渠道系统再调用外部厂商c系统。
(3)a渠道系统将结果返回给b业务系统。
这3个过程中，(2)过程，外部厂商的调用时是需要计费的。
当b业务系统并发量很高时，有100笔相同的三要素校验，由于是相同的三要素，a渠道只要调用一次厂商即可知道结果。那么a渠道系统如何控制不让100笔请求全部去访问外部厂商c系统呢？

小明提出了方案一：

在a系统中，
当100个线程同时请求过来，进行redis.setnx(“lock_key_phone&idno&name”,”demo”)，这样第一笔线程率先拿到锁，其他的线程等待，当thread(0)处理结束后，thread(0)进行delete(“lock_key_phone&idno&name”),把锁放开，thread(i)进行get(“lock_key_phone&idno&name”)拿到0，说明上一笔已经处理完成，这个时候，我们可以去查询上一笔的记录。

redisutils.setnx("lock_key_phone&idno&name","demo");
jsonobject result = a.request(b);
assetutils.notnull(result,responsecodeenum.success,"拿到结果");
resultdmo resultdmo = (resultdmo)beanutils.maptobean(result);
resultdao.insert(resultdmo);
if(result!=0){
  //上一笔同样的请求还未处理完成，轮训等待(具体如何轮训在此不展开)
}else{
  //上一笔同样的请求处理完成，进行查库操作
  resultdao.select("参数");
}

小宏说：小明的思想不严谨

问题：当100笔线程中一些线程超时或者系统宕机等意外情况发现，锁会一直被某些线程持有，造成死锁状态。
应该给缓存key设置一个超时时间。比如:200ms

redisutils.setnx("lock_key_phone&idno&name","demo",200);

这种情况是，大致判断了外部厂商c系统业务处理时间大概为200ms，

网上看还有一种方式（b）：

redisutils.setnx("lock_key_phone&idno&name",currenttime,200);
long old = redisutils.get("lock_key_phone&idno&name");
long new = system.currenttimemillis();
long time = new - old;
if(time>0){
//处理已经超时
redisutils.delete("lock_key_phone&idno&name");
}

（b）这种情况不严谨：当a获取setnx锁，a线程崩溃或超时，b、c线程同时get到old，且判断超时，可能出现b线程delete a线程的锁，并且setnx后；c线程又将b线程的锁delete,并且setnx。这种情况完全锁不住线程了。

（b）方案的升级版—->>（c）方案：

当a获取setnx锁，a线程崩溃或超时，b线程getset，获取old且判断超时，c线程getset，获取old（此时这个值是b刚刚set进去的），判断未超时，c继续等待。b线程delete a线程的锁，并且setnx后。这种情况是安全的。

需要注意的地方：
①不要轻易将get和getset混用，笔者认为getset单独使用比较好。
有一种情况，a、b、c、三个线程，a、b同时get，a立即返回了old，突然来了个c，卡在b之前getset了，且删除锁，那么b的get只能返回nil了。此时再根据时间戳对比：
a.get != (a.set)
b.get ! = (b.set)
这样a、b都没拿到锁，但是a其实已经获取到了锁。
②多个服务器时间的同步问题。

总结： 锁超时了该如何处理，通过getset方式判断时间戳差的方式，多比同时getset都得到超时，同时去setnx。总会有一个更快地去setnx。

二、通过incr抢占资源实现

1、incr

将 key 中储存的数字值增一。如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 incr 操作。如果值包含错误的类型，或字符串类型的值不能表示为数字，那么返回一个错误。

public static long incr(final string key) {
        return shardedclient.execute(new shardedjedisaction<long>() {
            @override
            public long doaction(shardedjedis shardedjedis) {
                shardedjedis.expire(key, 200);
                return shardedjedis.incr(key);
            }
        });
    }

还是上面的三要素的例子

long result = redisutils.incr("lock_key_phone&idno&name");
        if (result > 1) {
            //如果计数器>1,说明已经有请求进来
            throw new appexception(responsecode.fail.getcode(), "操作频繁");
        }


 jsonobject result = a.request(b);
 long endtime = system.currenttimemillis();
 long time = endtime - starttime;
  //如果处理时间大于incr的key存活时间,说明该笔请求已经超时
  if (time > 200) {
      //全局id,统计超时次数
      string key = "lock_key_phone&idno&name" + source;
      redisutils.incr(key);
      int total = integer.valueof(redisutils.get(key));
      //断言若超时10次,进行报警(报警不在次展开)
      assertutils.istrue(total < 10, responsecode.fail, "调用" + source + "渠道超时");
  }

这里设置了计数器的超时时间为200ms，如果请求超时，会有大量的线程同时访问，笔者这里有10笔同时过来，就启动报警。人为排查渠道。和setnx的不同是，某个线程超时，setnx的方式需要手动去判断，再去加锁，防止大量线程进入(这里可以通过轮训实现)；而incr的方式超时了，大量线程进来，我不做处理，但是这里的time>200是具有误差的。

到此这篇关于redis实现分布式锁(setnx、getset、incr)以及如何处理超时情况的文章就介绍到这了,更多相关redis setnx、getset、incr内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！