A. Redis的Setnx命令实现分布式锁
首先，分布式锁和我们平常讲到的锁原理基本一样，目的就是确保在多个线程并发时，只有一个线程在同一刻操作这个业务或者说方法、变量。
在一个进程中，也就是一个jvm或者说应用中，我们很容易去处理控制，在 java.util 并发包中已经为我们提供了这些方法去加锁，比如 synchronized 关键字或者 Lock 锁，都可以处理。
但是如果在分布式环境下，要保证多个线程同时只有1个能访问某个资源，就需要用到分布式锁。这里我们将介绍用Redis的 setnx 命令来实现分布式锁。
其实目前通常所说的 setnx 命令，并非单指redis的 setnx key value 这条命令，这条命令可能会在后期redis版本中删除。
一般代指redis中对 set 命令加上 nx 参数进行使用， set 这个命令，目前已经支持这么多参数可选：
从 Redis 2.6.12 版本开始， SET 命令的行为可以通过一系列参数来修改：
注入bean
这里同时启动5个线程并发往redis中存储 lock 这个key（key可以自定义，但需要一致），同时设置10秒的过期时间。
setIfAbsent 这个函数实现的功能与 setnx 命令一样，代表如果没有这个key则set成功获取到锁，否则set失败没有获取到锁。
获得锁后进行资源的操作，最后释放锁。
执行效果 ：
可以看到同时只有1个线程能够获取到锁。
使用 setnx 命令方式虽然操作比较简单方便，但是会有如下问题：
可以在丛肢再次获取锁时，如果锁被占用就get值，判断值是否是当前线程存的随机值，如果是则再次执行 set 命令重新上锁；当然为了保证原子性这些操作都要用 lua 脚本来执行。
可以使用 while 循环重复执行 setnx 命令，并设置一个超时时间退出循环。
可以尽量把锁自动过期的时间设的冗余一些。但也不能绝橡彻底解决。
可以在删除锁的时候先get值，判断值是否是当前线程存的随机值，只有相同才执行渗宏世删锁的操作；当然也要使用 lua 脚本执行来保证原子性。
分布式锁需要满足的特性
综上：使用 setnx 命令来实现分布式锁并不是一个很严谨的方案，如果是Java技术栈，我们可以使用 Redisson 库来解决以上问题，接下来的文章会介绍如何使用。
Redisson实现分布式锁
Redlock实现分布式锁
B. Springboot使用redis的setnx和getset实现并发锁、分布式锁
在日常开发中，很多业圆搏务场景必须保证原子性。举几个例子：
如果你只有一台服务器，只运行一个Java程序，那么可以使用Java语言自身的一些锁来实现原子性。但如果我们有多台服务器，甚至不同服务器上跑的是不同的语言。那这时候，我们就需要一个跨平台、跨语言的加锁方式。redis就是其中最方便的一种。
使用redis实现并发锁，主要是靠两个redis的命令：setnx和getset。
那我们的设计思路就是：
上面的代码使用了一个RedisService的类薯升，里面主要是简单封装了一数腔老下redis的操作，你可以替换为自己的service。代码如下：
以上代码有任何疑问，可以点击右侧边栏联系作者。收费5毛~交个朋友，欢迎来撩！
版权声明：《Springboot使用redis的setnx和getset实现并发锁、分布式锁》为CoderBBB作者“ʘᴗʘ”的原创文章，转载请附上原文出处链接及本声明。
原文链接：https://www.coderbbb.com/articles/2
C. java 链接redis 怎么加锁
我介绍一下Redis分布式锁吧：
一、定义redis实现分布式锁的接口
[java]viewplainprint?
packagecom.iol.common.util.concurrent.locks;
importjava.io.Serializable;
/**
*Description:定义redis实现分布式锁的算法<br/>
*_SMALL_TAIL.<br/>
*ProgramName:IOL_SMALL_TAIL<br/>
*Date:2015年11月8日
*
*@author王鑫
*@version1.0
*/
{
/**
*加锁算法<br/>
*@paramkey
*@return
*/
publicbooleanlock(Stringkey);
/**
*解锁算法<br/>
*@paramkey
*@return
*/
publicbooleanunLock(Stringkey);
}
二、redis分布式锁基础算法实现
[java]viewplainprint?
packagecom.iol.common.util.concurrent.locks.arithmetic;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importcom.iol.common.util.concurrent.locks.IRedisComponent;
importcom.iol.common.util.concurrent.locks.IRedisLockArithmetic;
/**
*Description:redis分布式锁基础算法实现<br/>
*_SMALL_TAIL.<br/>
*ProgramName:IOL_SMALL_TAIL<br/>
*Date:2015年11月9日
*
*@author王鑫
*@version1.0
*/
{
/**
*serialVersionUID
*/
=-8333946071502606883L;
privateLoggerlogger=LoggerFactory.getLogger(RedisLockBaseArithmetic.class);
/**
*redis操作方法
*/
;
/**
*超时时间，以毫秒为单位<br/>
*默认为5分钟
*/
privatelongovertime=5*60*1000L;
/**
*休眠时长，以毫秒为单位<br/>
*默认为100毫秒
*/
privatelongsleeptime=100L;
/**
*当前时间
*/
privatelongcurrentLockTime;
/**
*@
*/
publicvoidsetRedisComp(IRedisComponentredisComp){
this.redisComp=redisComp;
}
/**
*@paramovertimetheovertimetoset
*/
publicvoidsetOvertime(longovertime){
this.overtime=overtime;
}
/**
*@
*/
publicvoidsetSleeptime(longsleeptime){
this.sleeptime=sleeptime;
}
/*(non-Javadoc)
*@seecom.iol.common.util.concurrent.locks.IRedisLockArithmetic#lock(java.lang.String,java.lang.Long)
*/
@Override
publicbooleanlock(Stringkey){
while(true){
//当前加锁时间
currentLockTime=System.currentTimeMillis();
if(redisComp.setIfAbsent(key,currentLockTime)){
//获取锁成功
logger.debug("直接获取锁{key:{},currentLockTime:{}}",key,currentLockTime);
returntrue;
}else{
//其他线程占用了锁
logger.debug("检测到锁被占用{key:{},currentLockTime:{}}",key,currentLockTime);
LongotherLockTime=redisComp.get(key);
if(otherLockTime==null){
//其他系统释放了锁
//立刻重新尝试加锁
logger.debug("检测到锁被释放{key:{},currentLockTime:{}}",key,currentLockTime);
continue;
}else{
if(currentLockTime-otherLockTime>=overtime){
//锁超时
//尝试更新锁
logger.debug("检测到锁超时{key:{},currentLockTime:{},otherLockTime:{}}",key,currentLockTime,otherLockTime);
LongotherLockTime2=redisComp.getAndSet(key,currentLockTime);
if(otherLockTime2==null||otherLockTime.equals(otherLockTime2)){
logger.debug("获取到超时锁{key:{},currentLockTime:{},otherLockTime:{},otherLockTime2:{}}",key,currentLockTime,otherLockTime,otherLockTime2);
returntrue;
}else{
sleep();
//重新尝试加锁
logger.debug("重新尝试加锁{key:{},currentLockTime:{}}",key,currentLockTime);
continue;
}
}else{
//锁未超时
sleep();
//重新尝试加锁
logger.debug("重新尝试加锁{key:{},currentLockTime:{}}",key,currentLockTime);
continue;
}
}
}
}
}
/*(non-Javadoc)
*@seecom.iol.common.util.concurrent.locks.IRedisLockArithmetic#unLock(java.lang.String)
*/
@Override
publicbooleanunLock(Stringkey){
logger.debug("解锁{key:{}}",key);
redisComp.delete(key);
returntrue;
}
/**
*休眠<br/>
*@paramsleeptime
*/
privatevoidsleep(){
try{
Thread.sleep(sleeptime);
}catch(InterruptedExceptione){
thrownewLockException("线程异常中断",e);
}
}
}
D. 北京java课程分享分布式限流的运行原理
分布式编程架构技术我们在前几期的文章中已经给大家简单分析过很多次了，今天我们就一起来了解一下API网关分布式限流的运正滑悔行原理都有哪些。
API网关中针对一个API、API分组、接入应用APPID,IP等进行限流。这些限流条件都将会产生一个限流使用的key，在后续的限流中都是对这个key进行限流。
限流算法通常在API网关中可以采用令牌桶算法实现。
必须说明一点的是分布式限流由于有网络的开销，TPS的支持隔本地限流是有差距的，因此在对于TPS要求很高的场景，建议采用本地限流进行处理。
下面讨论我们应该采用redis的哪一种分布式锁的方案：
由于redis事务要得到锁的效果需要在高TPS时会产生大量的无效的访问请求，所以不建议在这种场景下使用。
SETNX/EX的锁方案会产生在过期时间的问题，同时也有异步复制master数据到slave的问题。相比lua方案会产生更多的不稳定性。
我建议采用lua的方案来实施分布式锁，因为都是单进程单线程的执行，因此在TPS上和二种方案没有大的区别，而且由于只是一个lua脚本在执行，甚至是可能纯lua执行可能会有更高的TPS。当然是lua脚本中可能还是会去设置过期时间，让闷但是应用server宕机并不会影响到redis中的锁。当然master异步复制的问题还是有，但是并不会造成问题，因为数据只会有1个lua脚本执行问题，下一个执行就正常了。
在实现方案的时候使用了Jedis库，北京java课程认为有一些问题举正在方案的实现层面我已经去做过验证了，可能也会是读者的疑问。
E. 使用Redisson实现分布式锁
Redisson的分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口，同时还支持自动过期解锁。
Redisson同时还为分布式锁提供了异步执行的相关方法：
Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。在提供了自动过期解锁坦历功能的同时，保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。
Redisson同时还为分布式可重入公平锁提供了异步执行的相关方法：
Redisson的RedissonMultiLock对象可以将多个RLock对象关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例。
Redisson的RedissonRedLock对象实现了 Redlock 介绍的加锁算法。该对象也可以用来将多个RLock
对象关联为一个红锁，每个RLock对象实例可以来自于不同的Redisson实例。
Redisson的分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。同时还支持自动过期解锁。该对象允许同时有多个读取锁，但是最多只能有一个写入锁。
Redisson的分布式信号量（Semaphore）Java对让颤搜象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。
Redisson的可过期性信号量（PermitExpirableSemaphore）实在RSemaphore对象的基础上，为每个信号增加了一个过期时间。每个信号可以通过独立洞禅的ID来辨识，释放时只能通过提交这个ID才能释放。
Redisson的分布式闭锁（CountDownLatch）Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。
Redisson 分布式锁和同步器
F. java怎么实现redis分布式锁
一基咐、使用分布式锁要满足的几个条件：
系统是一个分布式系统（关键是分布式，单机的可以使用ReentrantLock或者synchronized代码块来实现）
共享资源（各个系统访问同一个资源，资禅锋槐源的载体可能是传统关系型数据贺友库或者NoSQL）
同步访问（即有很多个进程同事访问同一个共享资源。没有同步访问，谁管你资源竞争不竞争）
二、应用的场景例子
管理后台的部署架构（多台tomcat服务器+redis【多台tomcat服务器访问一台redis】+mysql【多台tomcat服务器访问一台服务器上的mysql】）就满足使用分布式锁的条件。多台服务器要访问redis全局缓存的资源，如果不使用分布式锁就会出现问题。 看如下伪代码：
long N=0L;
//N从redis获取值
if(N<5){
N++；
//N写回redis
}
复制代码
G. Java 集群锁如何实现
1. 用数据库，在数据库建一张表，需要锁的节点都可以尝试用 select * from Lock where id=xx for update. 这个时候只有一个节点能拿到结果，其它的都会等待，就能实现一个简单的悲观锁。
2. 用 Zookeeper 来做分布式锁，具体可以搜一下。
3. 自己实现，搞个节点来做这个事情，所有要获取锁的都走 RPC 调用来请求锁，用完以后记得释放，不然其他的节点就会挂那里。
H. java trylock能实现分布式锁吗
一、zookeeper
1、实现原理：
基于zookeeper瞬时有序节点实现的分布式锁，其主要逻辑如下（该图来自于IBM网站）。大致思想即为：每个客户端对某个功能加锁时，在zookeeper上的与该功能对应的指定节点的目录下，生成一个唯一的瞬时有序节点。判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。
二、memcached分布式锁
1、实现原理：
memcached带有add函数，利用add函数的特性即可实现分布式锁。add和set的区别在于：如果多线程并发set，则每个set都会成功，但最后存储的值以最后的set的线程为准。而add的话则相反，add会添加第一个到达的值，并返回true，后续的添加则都会返回false。利用该点即可很轻松地实现分布式锁。
三、redis分布式锁
redis分布式锁即可以结合zk分布式锁锁高度安全和memcached并发场景下效率很好的优点，可以利用jedis客户端实现
I. 高并发没锁可不行，三种分布式锁详解
Java中的锁主要包括synchronized锁和JUC包中的锁，这些锁都是针对单个JVM实例上的锁，对于分布式环境如果我们需要加锁就显得无能为力。在单个JVM实例上，锁的竞争者通常是一些不同的线程，而在分布式环境中，锁的竞争者通常是一些不同的线程或者进程。如何实现在分布式环境中对一个对象进行加锁呢？答案就是分布式锁。
目前分布式锁的实现方案主要包括三种：
基于数据库实现分布式锁主要是利用数据库的唯一索引来实现，唯一索引天然具有排他性，这刚好符合我们对锁的要求：同一时刻只能允许一个竞争者获取锁。加锁时我们在数据库中插入一条锁记录，利用业务id进行防重。当第一个竞争者加锁成功后，第二个竞争者再来加锁就会抛出唯一索引冲突，如果抛出这个异常，我们就判定当前竞争者加锁失败。防重业务id需要我们自己来定义，例如我们的锁对象是一个方法，则我们的业务防重id就是这个方法的名字，如果锁定的对象是一个类，则业务防重id就是这个类名。
基于缓存实现分布式锁：理论上来说使用缓存来实现分布式锁的效率最高，加锁速度最快，因为Redis几乎都是纯内存操作，而基于数据库的方案和基于Zookeeper的方案都会涉及到磁盘文件IO，效率相对低下。一般使用Redis来实现分布式锁都是利用Redis的 SETNX key value 这个命令，只有当key不存在时才会执行成功，如果key已经存在则命令执行失败。
基于Zookeeper：Zookeeper一般用作配置中心，其实现分布式锁的原神如理和迟备Redis类似，我们在Zookeeper中创建瞬时节点，利用节点不能重复创建的特性来保证排他性。
在实现分布式锁的时候我们需要考虑一些问题，例如：分布式锁是否可重入，分布式锁的释放时机，分布式锁服务端是否有单点问题等。
上面已经分析了基于数据库实现分布式锁的基本原理：通过唯一索引保持排他性，加锁时插入一条记录，解锁是删除这条记录。下面我们就简要实现一下基于数据库的分布式锁。
id字段是数据库的自增id，unique_mutex字段就是我们的防重id，也就是加锁的对象，此对象唯一。在这张表上我们加了一个唯一索引，保证unique_mutex唯一性。holder_id代表竞争到锁的持有者id。
如果当前sql执行成功代表加锁成功，如果抛出唯一索引异常(DuplicatedKeyException)则代表加锁失败，当前锁已经被其他竞争者获取。
解锁很简单，直接删除此条记录即可。
是否可重入 ：就以上的方案来说，我们实现的分布式锁是不可重码瞎毁入的，即是是同一个竞争者，在获取锁后未释放锁之前再来加锁，一样会加锁失败，因此是不可重入的。解决不可重入问题也很简单：加锁时判断记录中是否存在unique_mutex的记录，如果存在且holder_id和当前竞争者id相同，则加锁成功。这样就可以解决不可重入问题。
锁释放时机 ：设想如果一个竞争者获取锁时候，进程挂了，此时distributed_lock表中的这条记录就会一直存在，其他竞争者无法加锁。为了解决这个问题，每次加锁之前我们先判断已经存在的记录的创建时间和当前系统时间之间的差是否已经超过超时时间，如果已经超过则先删除这条记录，再插入新的记录。另外在解锁时，必须是锁的持有者来解锁，其他竞争者无法解锁。这点可以通过holder_id字段来判定。
数据库单点问题 ：单个数据库容易产生单点问题：如果数据库挂了，我们的锁服务就挂了。对于这个问题，可以考虑实现数据库的高可用方案，例如MySQL的MHA高可用解决方案。
使用Jedis来和Redis通信。
可以看到，我们加锁就一行代码：
jedis.set(String key, String value, String nxxx, String expx, int time);
这个set()方法一共五个形参：
第一个为key,我们使用key来当锁，因为key是唯一的。
第二个为value，这里写的是锁竞争者的id，在解锁时，我们需要判断当前解锁的竞争者id是否为锁持有者。
第三个为nxxx，这个参数我们填的是NX，意思是SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作。
第四个为expx，这个参数我们传的是PX，意思是我们要给这个key加一个过期时间的设置，具体时间由第五个参数决定；
第五个参数为time，与第四个参数相呼应，代表key的过期时间。
总的来说，执行上面的set()方法就只会导致两种结果：1.当前没有锁（key不存在）,那么久进行加锁操作，并对锁设置一个有效期，同时value表示加锁的客户端。2.已经有锁存在，不做任何操作。
上述解锁请求中， SET_IF_NOT_EXIST (不存在则执行)保证了加锁请求的排他性，缓存超时机制保证了即使一个竞争者加锁之后挂了，也不会产生死锁问题：超时之后其他竞争者依然可以获取锁。通过设置value为竞争者的id，保证了只有锁的持有者才能来解锁，否则任何竞争者都能解锁，那岂不是乱套了。
解锁的步骤：
注意到这里解锁其实是分为2个步骤，涉及到解锁操作的一个原子性操作问题。这也是为什么我们解锁的时候用Lua脚本来实现，因为Lua脚本可以保证操作的原子性。那么这里为什么需要保证这两个步骤的操作是原子操作呢？
设想：假设当前锁的持有者是竞争者1，竞争者1来解锁，成功执行第1步，判断自己就是锁持有者，这是还未执行第2步。这是锁过期了，然后竞争者2对这个key进行了加锁。加锁完成后，竞争者1又来执行第2步，此时错误产生了：竞争者1解锁了不属于自己持有的锁。可能会有人问为什么竞争者1执行完第1步之后突然停止了呢？这个问题其实很好回答，例如竞争者1所在的JVM发生了GC停顿，导致竞争者1的线程停顿。这样的情况发生的概率很低，但是请记住即使只有万分之一的概率，在线上环境中完全可能发生。因此必须保证这两个步骤的操作是原子操作。
是否可重入 ：以上实现的锁是不可重入的，如果需要实现可重入，在 SET_IF_NOT_EXIST 之后，再判断key对应的value是否为当前竞争者id，如果是返回加锁成功，否则失败。
锁释放时机 ：加锁时我们设置了key的超时，当超时后，如果还未解锁，则自动删除key达到解锁的目的。如果一个竞争者获取锁之后挂了，我们的锁服务最多也就在超时时间的这段时间之内不可用。
Redis单点问题 ：如果需要保证锁服务的高可用，可以对Redis做高可用方案：Redis集群+主从切换。目前都有比较成熟的解决方案。
利用Zookeeper创建临时有序节点来实现分布式锁：
其基本思想类似于AQS中的等待队列，将请求排队处理。其流程图如下：
解决不可重入 ：客户端加锁时将主机和线程信息写入锁中，下一次再来加锁时直接和序列最小的节点对比，如果相同，则加锁成功，锁重入。
锁释放时机 ：由于我们创建的节点是顺序临时节点，当客户端获取锁成功之后突然session会话断开，ZK会自动删除这个临时节点。
单点问题 ：ZK是集群部署的，主要一半以上的机器存活，就可以保证服务可用性。
Zookeeper第三方客户端curator中已经实现了基于Zookeeper的分布式锁。利用curator加锁和解锁的代码如下：
J. java怎么实现redis分布式锁
Redis有一系列的命令，特点是以NX结尾，NX是Not eXists的缩写，如SETNX命令就应该理解为：SET if Not eXists。这系列的命令非常有用，这里讲使用SETNX来实现分布式锁。
用SETNX实现分布式锁
利用SETNX非常简单地实现分布式锁。例如：某客户端要获得一个名字foo的锁，客户端使用下面的命令进行获取：
SETNX lock.foo <current Unix time + lock timeout + 1>
如返回1，则该客户端获得锁，把lock.foo的键值设置为时间值表示该键已被锁定，该客户端最后可以通过DEL lock.foo来释放该锁。
如返回0，表明该锁已被其他客户端取得，这时我们可以先返回或进行重试等对方完成或等待锁超时。
解决死锁
上面的锁定逻辑有一个问题：如果一个持有锁的客户端失败或崩溃了不能释放锁，该怎么解决？我们可以通过锁的键对应的时间戳来判断这种情况是否发生了，如果当前的时间已经大于lock.foo的值，说明该锁已失效，可以被重新使用。
发生这种情况时，可不能简单的通过DEL来删除锁，然后再SETNX一次，当多个客户端检测到锁超时后都会尝试去释放它，这里就可能出现一个竞态条件,让我们模拟一下这个场景：
C0操作超时了，但它还持有着锁，C1和C2读取lock.foo检查时间戳，先后发现超时了。
C1 发送DEL lock.foo
C1 发送SETNX lock.foo 并且成功了。
C2 发送DEL lock.foo
C2 发送SETNX lock.foo 并且成功了。
这样一来，C1，C2都拿到了锁！问题大了！
幸好这种问题是可以避免D，让我们来看看C3这个客户端是怎样做的：
C3发送SETNX lock.foo 想要获得锁，由于C0还持有锁，所以Redis返回给C3一个0
C3发送GET lock.foo 以检查锁是否超时了，如果没超时，则等待或重试。
反之，如果已超时，C3通过下面的操作来尝试获得锁：
GETSET lock.foo <current Unix time + lock timeout + 1>
通过GETSET，C3拿到的时间戳如果仍然是超时的，那就说明，C3如愿以偿拿到锁了。
如果在C3之前，有个叫C4的客户端比C3快一步执行了上面的操作，那么C3拿到的时间戳是个未超时的值，这时，C3没有如期获得锁，需要再次等待或重试。留意一下，尽管C3没拿到锁，但它改写了C4设置的锁的超时值，不过这一点非常微小的误差带来的影响可以忽略不计。
注意：为了让分布式锁的算法更稳键些，持有锁的客户端在解锁之前应该再检查一次自己的锁是否已经超时，再去做DEL操作，因为可能客户端因为某个耗时的操作而挂起，操作完的时候锁因为超时已经被别人获得，这时就不必解锁了。
示例伪代码
根据上面的代码，我写了一小段Fake代码来描述使用分布式锁的全过程：
# get lock
lock = 0
while lock != 1:
timestamp = current Unix time + lock timeout + 1
lock = SETNX lock.foo timestamp
if lock == 1 or (now() > (GET lock.foo) and now() > (GETSET lock.foo timestamp)):
break;
else:
sleep(10ms)
# do your job
do_job()
# release
if now() < GET lock.foo:
DEL lock.foo
是的，要想这段逻辑可以重用，使用python的你马上就想到了Decorator，而用Java的你是不是也想到了那谁？AOP + annotation？行，怎样舒服怎样用吧，别重复代码就行。