分布式锁其实可以理解为：控制分布式系统有序的去对共享资源进行操作，通过互斥来保持一致性。 举个不太恰当的例子：假设共享的资源就是一个房子，里面有各种书，分布式系统就是要进屋看书的人，分布式锁就是保证这个房子只有一个门并且一次只有一个人可以进，而且门只有一把钥匙。然后许多人要去看书，可以，排队，第一个人拿着钥匙把门打开进屋看书并且把门锁上，然后第二个人没有钥匙，那就等着，等第一个出来，然后你在拿着钥匙进去，然后就是以此类推

1、锁的分类

1.1、线程锁

主要用来给方法、代码块加锁。当某个方法或代码使用锁，在同一时刻仅有一个线程执行该方法或该代码段。线程锁只在同一JVM中有效果，因为线程锁的实现在根本上是依靠线程之间共享内存实现的，比如synchronized是共享对象头，显示锁Lock是共享某个变量(state)。

1.2、进程锁

为了控制同一操作系统中多个进程访问某个共享资源，因为进程具有独立性，各个进程无法访问其他进程的资源，因此无法通过synchronized等线程锁实现进程锁。

1.3、分布式锁

当多个进程不在同一个系统中，用分布式锁控制多个进程对资源的访问。

2、分布式锁应具备什么条件

1. 互斥性：任意时刻，只能有一个客户端获取锁，不能同时有两个客户端获取到锁。
2. 安全性：锁只能被持有该锁的客户端删除，不能由其它客户端删除。
3. 死锁：获取锁的客户端因为某些原因(如down机等)而未能释放锁，其它客户端再也无法获取到该锁。
4. 容错：当部分节点(redis节点等)down机时，客户端仍然能够获取锁和释放锁。

3、分布式锁的核心

1. 加锁：要保证同一时间只有一个客户端可以拿到锁，得到锁之后开始执行相关业务。
2. 解锁：在业务执行完毕后，须及时释放锁，以便其他线程可以进入。
3. 锁超时：如果一个得到锁的线程在执行任务的过程中挂掉，来不及显式地释放锁，这块资源将会永远被锁住（死锁），别的线程再也别想进来。所以需要设置一个超时时间，以保证即使没有被显式释放，这把锁也要在一定时间后自动释放，避免死锁。

4、分布式锁的实现方式

4.1、基于数据库实现

4.1.1、基于数据库表实现

在数据库中创建一个表，表中包含方法名等字段，并在方法名字段上创建唯一索引，想要执行某个方法，就使用这个方法名向表中插入数据，成功插入则获取锁，执行完成后删除对应的行数据释放锁。
（1）创建如下数据表

DROP TABLE IF EXISTS `method_lock`;
CREATE TABLE `method_lock` (`id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',`method_name` varchar(64) NOT NULL COMMENT '锁定的方法名',`desc` varchar(255) NOT NULL COMMENT '备注信息',`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`),UNIQUE KEY `uidx_method_name` (`method_name`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 COMMENT='锁定中的方法';

（2）执行方法

想要执行某个方法，就调用这个方法向数据表method_lock中插入数据：

INSERT INTO method_lock (method_name, desc) 
VALUES ('methodName', '测试的methodName');

 （3）锁释放

成功插入则表示获取到锁，插入失败则表示获取锁失败；插入成功后，就好继续方法体的内容，执行完成后删除对应的行数据释放锁：

delete from method_lock where method_name ='methodName';

  （4）缺点

• 这把锁强依赖数据库的可用性，数据库是一个单点，一旦数据库挂掉，会导致业务系统不可用。
• 这把锁没有失效时间，一旦解锁操作失败，就会导致锁记录一直在数据库中，其他线程无法再获得到锁。
• 这把锁只能是非阻塞的，因为数据的insert操作，一旦插入失败就会直接报错。没有获得锁的线程并不会进入排队队列，要想再次获得锁就要再次触发获得锁操作。
• 这把锁是非重入的，同一个线程在没有释放锁之前无法再次获得该锁。因为数据中数据已经存在了。
• 这把锁是非公平锁，所有等待锁的线程凭运气去争夺锁。

   （5）调优

1. 因为是基于数据库实现的，数据库的可用性和性能将直接影响分布式锁的可用性及性能，所以，数据库需要双机部署、数据同步、主备切换；
2. 不具备可重入的特性，因为同一个线程在释放锁之前，行数据一直存在，无法再次成功插入数据，所以，需要在表中新增一列，用于记录当前获取到锁的机器和线程信息，在再次获取锁的时候，先查询表中机器和线程信息是否和当前机器和线程相同，若相同则直接获取锁；
3. 没有锁失效机制，因为有可能出现成功插入数据后，服务器宕机了，对应的数据没有被删除，当服务恢复后一直获取不到锁，所以，需要在表中新增一列，用于记录失效时间，并且需要有定时任务清除这些失效的数据；
4. 不具备阻塞锁特性，获取不到锁直接返回失败，所以需要优化获取逻辑，循环多次去获取。
5. 在实施的过程中会遇到各种不同的问题，为了解决这些问题，实现方式将会越来越复杂；依赖数据库需要一定的资源开销，性能问题需要考虑。

4.1.2、基于数据库排他锁实现

利用我们的创建method_lock表，通过数据库的排他锁来实现分布式锁。 基于MySql的InnoDB引擎，可以使用以下方法来实现加锁操作：

"select * from methodLock where method_name= '" + methodName + "' for update";  //悲观锁

public boolean lock(){connection.setAutoCommit(false)while(true){try{result = select * from methodLock where method_name=xxx for update;if(result==null){return true;}}catch(Exception e){}sleep(1000);}return false;
}

        在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁。当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。我们可以认为获得排它锁的线程即可获得分布式锁，当获取到锁之后，可以执行方法的业务逻辑，执行完方法之后，再通过以下方法解锁：

public void unlock(){connection.commit();
}

通过connection.commit();操作来释放锁。这种方法可以有效的解决上面提到的无法释放锁和阻塞锁的问题。for update语句会在执行成功后立即返回，在执行失败时一直处于阻塞状态，直到成功。锁定之后服务宕机，无法释放？使用这种方式，服务宕机之后数据库会自己把锁释放掉。

4.2、基于zookeeper实现分布式锁

基于zookeeper临时有序节点可以实现的分布式锁。ZooKeeper是一个为分布式应用提供一致性服务的开源组件，它内部是一个分层的文件系统目录树结构，规定同一个目录下只能有一个唯一文件名。Zookeeper的数据存储结构就像一棵树，这棵树由节点组成，这种节点叫做Znode。Znode分为四种类型：

1.持久节点 （PERSISTENT）

        默认的节点类型。创建节点的客户端与zookeeper断开连接后，该节点依旧存在 。

2.持久节点顺序节点（PERSISTENT_SEQUENTIAL）

        所谓顺序节点，就是在创建节点时，Zookeeper根据创建的时间顺序给该节点名称进行编号：

3.临时节点（EPHEMERAL）

        和持久节点相反，当创建节点的客户端与zookeeper断开连接后，临时节点会被删除：

4.临时顺序节点（EPHEMERAL_SEQUENTIAL）

        顾名思义，临时顺序节点结合和临时节点和顺序节点的特点：在创建节点时，Zookeeper根据创建的时间顺序给该节点名称进行编号；当创建节点的客户端与zookeeper断开连接后，临时节点会被删除。

基于ZooKeeper实现分布式锁的步骤如下：
（1）创建一个目录mylock；
（2）线程A想获取锁就在mylock目录下创建临时顺序节点；
（3）获取mylock目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁；
（4）线程B获取所有节点，判断自己不是最小节点，设置监听比自己次小的节点；
（5）线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是不是最小的节点，如果是则获得锁。

原文链接：https://www.cnblogs.com/liuqingzheng/p/11080501.html

大致思想即为：每个客户端对某个方法加锁时，在zookeeper上的与该方法对应的指定节点的目录下，生成一个唯一的瞬时有序节点。 判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。 当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。

原文链接：https://blog.csdn.net/u010963948/article/details/79006572

使用zookeeper能解决的分布式问题：

锁无法释放？
    使用Zookeeper可以有效的解决锁无法释放的问题，因为在创建锁的时候，客户端会在ZK中创建一个临时节点，一旦客户端获取到锁之后突然挂掉（Session连接断开），那么这个临时节点就会自动删除掉。其他客户端就可以再次获得锁。

非阻塞锁？
    使用Zookeeper可以实现阻塞的锁，客户端可以通过在ZK中创建顺序节点，并且在节点上绑定监听器，一旦节点有变化，Zookeeper会通知客户端，客户端可以检查自己创建的节点是不是当前所有节点中序号最小的，如果是，那么自己就获取到锁，便可以执行业务逻辑了。

不可重入？
    使用Zookeeper也可以有效的解决不可重入的问题，客户端在创建节点的时候，把当前客户端的主机信息和线程信息直接写入到节点中，下次想要获取锁的时候和当前最小的节点中的数据比对一下就可以了。如果和自己的信息一样，那么自己直接获取到锁，如果不一样就再创建一个临时的顺序节点，参与排队。

单点问题？
    使用Zookeeper可以有效的解决单点问题，ZK是集群部署的，只要集群中有半数以上的机器存活，就可以对外提供服务。

公平问题？
    使用Zookeeper可以解决公平锁问题，客户端在ZK中创建的临时节点是有序的，每次锁被释放时，ZK可以通知最小节点来获取锁，保证了公平。

Zookeeper数据同步问题：
        Zookeeper是一个保证了弱一致性即最终一致性的分布式组件。Zookeeper采用称为Quorum Based Protocol的数据同步协议。假如Zookeeper集群有N台Zookeeper服务器(N通常取奇数，3台能够满足数据可靠性同时有很高读写性能，5台在数据可靠性和读写性能方面平衡最好)，那么用户的一个写操作，首先同步到N/2 + 1台服务器上，然后返回给用户，提示用户写成功。基于Quorum Based Protocol的数据同步协议决定了Zookeeper能够支持什么强度的一致性。

        在分布式环境下，满足强一致性的数据储存基本不存在，它要求在更新一个节点的数据，需要同步更新所有的节点。这种同步策略出现在主从同步复制的数据库中。但是这种同步策略，对写性能的影响太大而很少见于实践。因为Zookeeper是同步写N/2+1个节点，还有N/2个节点没有同步更新，所以Zookeeper不是强一致性的。

        用户的数据更新操作，不保证后续的读操作能够读到更新后的值，但是最终会呈现一致性。牺牲一致性，并不是完全不管数据的一致性，否则数据是混乱的，那么系统可用性再高分布式再好也没有了价值。牺牲一致性，只是不再要求关系型数据库中的强一致性，而是只要系统能达到最终一致性即可。
        推荐一个Apache的开源库Curator（https://github.com/apache/curator/），它是一个ZooKeeper客户端，Curator提供的InterProcessMutex是分布式锁的实现，acquire方法用于获取锁，release方法用于释放锁。

Zookeeper实现分布式的优缺点：
优点：具备高可用、可重入、阻塞锁特性，可解决失效死锁问题。

缺点：因为需要频繁的创建和删除节点，性能上不如Redis方式。

4.3、基于缓存实现分布式锁

4.3.1、使用命令

（1）加锁：
    SETNX key value：当且仅当key不存在时，set一个key为value的字符串，返回1；若key存在，则什么都不做，返回0。

（2）设置锁的超时：
    expire key timeout：为key设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁。

（3）释放锁：
    del key：删除key
    在使用Redis实现分布式锁的时候，主要就会使用到这三个命令。除此之外，我们还可以使用set key value NX EX max-lock-time 实现加锁，并且使用 EVAL 命令执行lua脚本实现解锁。

    EX seconds ： 将键的过期时间设置为 seconds 秒。 执行 SET key value EX seconds 的效果等同于执行 SETEX key seconds value 。

    PX milliseconds ： 将键的过期时间设置为 milliseconds 毫秒。 执行 SET key value PX milliseconds 的效果等同于执行 PSETEX key milliseconds value 。

    NX ： 只在键不存在时， 才对键进行设置操作。 执行 SET key value NX 的效果等同于执行 SETNX key value 。

    XX ： 只在键已经存在时， 才对键进行设置操作。

4.3.2、实现思路

（1）获取锁的时候，使用setnx加锁，并使用expire命令为锁添加一个超时时间，超过该时间则自动释放锁，锁的value值为一个随机生成的UUID，通过此在释放锁的时候进行判断。

（2）获取锁的时候还设置一个获取的超时时间，若超过这个时间则放弃获取锁。

（3）释放锁的时候，通过UUID判断是不是该锁，若是该锁，则执行delete进行锁释放。

import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;import java.util.Collections;public class RedisTool2 {private static Jedis jedis = new Jedis("127.0.0.1",6379);private static final String LOCK_SUCCESS = "OK";private static final String SET_IF_NOT_EXIST = "NX";private static final String SET_WITH_EXPIRE_TIME = "PX";private static final Long RELEASE_SUCCESS = 1L;/* EX seconds ： 将键的过期时间设置为 seconds 秒。 执行 SET key value EX seconds 的效果等同于执行 SETEX key seconds value 。 PX milliseconds ： 将键的过期时间设置为 milliseconds 毫秒。 执行 SET key value PX milliseconds 的效果等同于执行 PSETEX key milliseconds value 。 NX ： 只在键不存在时， 才对键进行设置操作。 执行 SET key value NX 的效果等同于执行 SETNX key value 。 XX ： 只在键已经存在时， 才对键进行设置操作。*//* 尝试获取分布式锁* @param lockKey 锁* @param requestId 请求标识* @param expireTime 超期时间(过期时间) 需要根据实际的业务场景确定* @return 是否获取成功*/public static boolean tryGetDistributedLock(String lockKey, String requestId, int expireTime) {SetParams params = new SetParams();String result = jedis.set(lockKey, requestId, params.nx().ex(expireTime));if (LOCK_SUCCESS.equals(result)) {return true;}return false;}/* 尝试获取分布式锁* @param lockKey 锁* @param requestId 请求标识* @param expireTime 超期时间（过期时间）需要根据实际的业务场景确定* @return 是否获取成功*/public static boolean tryGetDistributedLock1(String lockKey, String requestId, int expireTime){//只在键 key 不存在的情况下， 将键 key 的值设置为 value 。若键 key 已经存在， 则 SETNX 命令不做任何动作。设置成功返回1，失败返回0long code = jedis.setnx(lockKey, requestId);   //保证加锁的原子操作//通过timeOut设置过期时间保证不会出现死锁【避免死锁】jedis.expire(lockKey, expireTime);   //设置键的过期时间if(code == 1){return true;}return false;}/* 解锁操作* @param key 锁标识* @param value 客户端标识* @return*/public static Boolean unLock(String key,String value){//luaScript 这个字符串是个lua脚本，代表的意思是如果根据key拿到的value跟传入的value相同就执行del，否则就返回0【保证安全性】String luaScript = "if redis.call(\\"get\\",KEYS[1]) == ARGV[1] then return redis.call(\\"del\\",KEYS[1]) else  return 0 end";//jedis.eval(String,list,list);这个命令就是去执行lua脚本，KEYS的集合就是第二个参数，ARGV的集合就是第三参数【保证解锁的原子操作】Object var2 = jedis.eval(luaScript, Collections.singletonList(key), Collections.singletonList(value));if (RELEASE_SUCCESS == var2) {return true;}return false;}/* 解锁操作* @param key  锁标识* @param value  客户端标识* @return*/public static Boolean unLock1(String key, String value){//key就是redis的key值作为锁的标识，value在这里作为客户端的标识，只有key-value都比配才有删除锁的权利【保证安全性】String oldValue = jedis.get(key);long delCount = 0;  //被删除的key的数量if(oldValue.equals(value)){delCount = jedis.del(key);}if(delCount > 0){  //被删除的key的数量大于0，表示删除成功return true;}return false;}/* 重试机制：* 如果在业务中去拿锁如果没有拿到是应该阻塞着一直等待还是直接返回，这个问题其实可以写一个重试机制，* 根据重试次数和重试时间做一个循环去拿锁，当然这个重试的次数和时间设多少合适，是需要根据自身业务去衡量的* @param key 锁标识* @param value 客户端标识* @param timeOut 过期时间* @param retry 重试次数* @param sleepTime 重试间隔时间* @return*/public Boolean lockRetry(String key,String value,int timeOut,Integer retry,Long sleepTime){Boolean flag = false;try {for (int i=0;i<retry;i++){flag = tryGetDistributedLock(key,value,timeOut);if(flag){break;}Thread.sleep(sleepTime);}}catch (Exception e){e.printStackTrace();}return flag;}}

5、参考文章

参考文章

6、示例代码

SpinLock类：

package com.yibin.blnp.redis;import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;/* 用途：自旋锁 @version v1.01* @Author liaoyibin 2045165565@qq.com* @createDate 2023/3/16 21:12* @modifyRecord <pre>* 版本 修改人 修改时间 修改内容描述* ----------------------------------------------* 1.00 liaoyibin 2023/3/16 21:12 新建* ----------------------------------------------* </pre>*/
public class SpinLock {/* 成功锁标志/private static final String LOCK_SUCCESS = "OK";/* 失败锁标识/private static final long UNLOCK_SUCCESS = 1L;/* @author liaoyibin* 描述： 尝试获取分布式锁* @Date 21:19 2023/3/16* @param jedis Redis客户端* @param lockKey 锁键值* @param value 锁的值* @param expireTime 超期时间* @return boolean 是否获取成功/public static boolean tryLock(Jedis jedis, String lockKey, String value, int expireTime) {// 自旋锁while (true) {// set key value ex seconds nx(只有键不存在的时候才会设置key)String result = jedis.set(lockKey, value,SetParams.setParams().ex(expireTime).nx());if (LOCK_SUCCESS.equals(result)) {return true;}}}/* @author liaoyibin* 描述： 释放分布式锁* @Date 21:26 2023/3/16* @param jedis   Redis客户端* @param lockKey 锁* @return boolean 是否释放成功/public static boolean unlock(Jedis jedis, String lockKey) {Long result = jedis.del(lockKey);if (UNLOCK_SUCCESS == result) {return true;}return false;}
}

SpinLockTest类：

package com.yibin.blnp.redis;import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;import java.util.UUID;/* 用途：分布式自旋锁测试 @version v1.01* @Author liaoyibin 2045165565@qq.com* @createDate 2023/3/16 21:40* @modifyRecord <pre>* 版本 修改人 修改时间 修改内容描述* ----------------------------------------------* 1.00 liaoyibin 2023/3/16 21:40 新建* ----------------------------------------------* </pre>*/
public class SpinLockTest {/* 初始次数/private int count = 0;/* 加锁 key/private String lockKey = "lock";/* @author liaoyibin* 描述： 加锁* @Date 21:59 2023/3/16* @param jedis * @return void/private void addLock(Jedis jedis) {// 加锁boolean locked = SpinLock.tryLock(jedis, lockKey, UUID.randomUUID().toString(), 60);try {if (locked) {for (int i = 0; i < 500; i++) {count++;}}} catch (Exception e) {e.printStackTrace();} finally {SpinLock.unlock(jedis, lockKey);}}public static void main(String[] args) throws Exception {SpinLockTest redisLockTest = new SpinLockTest();JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();jedisPoolConfig.setMinIdle(1);jedisPoolConfig.setMaxTotal(5);JedisPool jedisPool = new JedisPool(jedisPoolConfig,"192.168.56.111", 6379, 1000, "123456");Thread t1 = new Thread(() -> redisLockTest.addLock(jedisPool.getResource()));Thread t2 = new Thread(() -> redisLockTest.addLock(jedisPool.getResource()));t1.start();t2.start();t1.join();t2.join();System.out.println(redisLockTest.count);}
}