什么时候用分布式锁（终于懂什么是分布式锁）

模拟一个秒杀接口：

商品表：

单机情况下，用Jmeter发送1000个请求过来：

由于加了sychronized进行方法同步，结果正常。

现在模拟集群环境，还是用上面的接口，但启动两个服务，分别是8080和8081端口，用nginx负载均衡到两个tomcat，用Jmeter发送1000个请求到nginx：

发现库存并没有-1000，并且控制的库存量打印有重复。

结论:我们在系统中修改已有数据时，需要先读取，然后进行修改保存，此时很容易遇到并发问题。由于修改和保存不是原子操作，在并发场景下，部分对数据的操作可能会丢失。在单服务器系统我们常用本地锁来避免并发带来的问题，然而，当服务采用集群方式部署时，本地锁无法在多个服务器之间生效，这时候保证数据的一致性就需要分布式锁来实现。

基于数据库的分布式锁, 常用的一种方式是使用表的唯一约束特性。当往数据库中成功插入一条数据时, 代表只获取到锁。将这条数据从数据库中删除，则释放锁。

CREATE TABLE `database_lock` ( `id` BIGINT NOT NULL AUTO_INCREMENT, `resource` varchar(1024) NOT NULL DEFAULT "" COMMENT '资源', `lock_id` varchar(1024) NOT NULL DEFAULT "" COMMENT '唯一锁编码'， `count` int(11) NOT NULL DEFAULT '0' COMMENT '锁的次数，可重入锁', PRIMARY KEY (`id`), UNIQUE KEY `uiq_lock_id` (`lock_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据库分布式锁表';

当我们想要获得锁时，可以插入一条数据INSERT INTO database_lock(resource,lock_id,count) VALUES ("resource","lock_id",1);

注意：在表database_lock中，lock_id字段做了唯一性约束，可以是机器的mac地址 线程编号,这样如果有多个请求同时提交到数据库的话，数据库可以保证只有一个操作可以成功（其它的会报错：ERROR 1062 (23000): Duplicate entry ‘1’ for key ‘uiq_lock_id’），那么我们就可以认为操作成功的那个请求获得了锁。

当需要释放锁的时，可以删除这条数据：DELETE FROM database_lock where method_name ='resource' and cust_id = 'lock_id'

可重入锁:UPDATE database_lock SET count = count 1 WHERE method_name ='resource' AND cust_id = 'lock_id'

伪代码:

public void test(){ String resource = "resource"; String lock_id = "lock_id"; if(!checkReentrantLock(resource,lock_id)){ lock(resource,lock_id);//加锁 }else{ reentrantLock(resource,lock_id); //可重入锁 1 } //业务处理 unlock(resource,lock_id);//释放锁 }

这种实现方式非常的简单，但是需要注意以下几点：

1. 这种锁没有失效时间，一旦释放锁的操作失败就会导致锁记录一直在数据库中，其它线程无法获得锁。这个缺陷也很好解决，比如可以做一个定时任务去定时清理。
2. 这种锁的可靠性依赖于数据库。建议设置备库，避免单点，进一步提高可靠性。
3. 这种锁是非阻塞的，因为插入数据失败之后会直接报错，想要获得锁就需要再次操作。如果需要阻塞式的，可以弄个for循环、while循环之类的，直至INSERT成功再返回。
4. 这种锁也是非可重入的，因为同一个线程在没有释放锁之前无法再次获得锁，因为数据库中已经存在同一份记录了。想要实现可重入锁，可以在数据库中添加一些字段，比如获得锁的主机信息、线程信息等，那么在再次获得锁的时候可以先查询数据，如果当前的主机信息和线程信息等能被查到的话，可以直接把锁分配给它。

Redis 锁主要利用 Redis 的 setnx 命令。

• 加锁命令：SETNX key value，当键不存在时，对键进行设置操作并返回成功，否则返回失败。KEY 是锁的唯一标识，一般按业务来决定命名。
• 解锁命令：DEL key，通过删除键值对释放锁，以便其他线程可以通过 SETNX 命令来获取锁。
• 锁超时：EXPIRE key timeout, 设置 key 的超时时间，以保证即使锁没有被显式释放，锁也可以在一定时间后自动释放，避免资源被永远锁住。

if (setnx(key, 1) == 1){ expire(key, 30) try { //TODO 业务逻辑 } finally { del(key) } }

使用SpingBoot集成Redis后使用分布式锁：

可以看到打印的日志不再有重复的库存量，最小的库存量与数据库中的一致。

Redis分布式锁可能存在一些问题：1.设置过期时间A客户端获取锁成功，但是在释放锁之前崩溃了，此时该客户端实际上已经失去了对公共资源的操作权，但却没有办法请求解锁（删除 Key-Value 键值对），那么，它就会一直持有这个锁，而其它客户端永远无法获得锁。在加锁时为锁设置过期时间，当过期时间到达，Redis 会自动删除对应的 Key-Value，从而避免死锁。2.SETNX 和 EXPIRE 非原子性如果SETNX成功，在设置锁超时时间之前，服务器挂掉、重启或网络问题等，导致EXPIRE命令没有执行，锁没有设置超时时间变成死锁。Redis 2.6.12 之后 Redis 支持 nx 和 ex 操作是同一原子操作。3.锁误解除如果线程 A 成功获取到了锁，并且设置了过期时间 30 秒，但线程 A 执行时间超过了 30 秒，锁过期自动释放，此时线程 B 获取到了锁；随后 A 执行完成，线程 A 使用 DEL 命令来释放锁，但此时线程 B 加的锁还没有执行完成，线程 A 实际释放的线程 B 加的锁。通过在 value 中设置当前线程加锁的标识，在删除之前验证 key 对应的 value 判断锁是否是当前线程持有。可生成一个 UUID 标识当前线程

4.超时解锁导致并发如果线程 A 成功获取锁并设置过期时间 30 秒，但线程 A 执行时间超过了 30 秒，锁过期自动释放，此时线程 B 获取到了锁，线程 A 和线程 B 并发执行。一般有两种方式解决该问题：将过期时间设置足够长，确保代码逻辑在锁释放之前能够执行完成。为获取锁的线程增加守护线程，为将要过期但未释放的锁增加有效时间。更好的方法是是使用Redission，WatchDog机制会为将要过期但未释放的锁增加有效时间。

5.redis主从复制A客户端在Redis的master节点上拿到了锁，但是这个加锁的key还没有同步到slave节点，master故障，发生故障转移，一个slave节点升级为master节点，B客户端也可以获取同个key的锁，但客户端A也已经拿到锁了，这就导致多个客户端都拿到锁。使用RedLock

1. 首先生成多个Redis集群的Rlock，并将其构造成RedLock。
2. 如果循环加锁的过程中加锁失败，那么需要判断加锁失败的次数是否超出了最大值，这里的最大值是根据集群的个数，比如三个那么只允许失败一个，五个的话只允许失败两个，要保证多数成功。
3. 加锁的过程中需要判断是否加锁超时，有可能我们设置加锁只能用3ms，第一个集群加锁已经消耗了3ms了。那么也算加锁失败。
4. 2，3步里面加锁失败的话，那么就会进行解锁操作，解锁会对所有的集群在请求一次解锁。

可以看见RedLock基本原理是利用多个Redis集群，用多数的集群加锁成功，减少Redis某个集群出故障，造成分布式锁出现问题的概率。

1.多个客户端创建一个锁节点下的一个接一个的临时顺序节点2.如果自己是第一个临时顺序节点，那么这个客户端加锁成功；如果自己不是第一个节点，就对自己上一个节点加监听器3.当某个客户端监听到上一个节点释放锁，自己就排到前面去了，此时继续执行步骤2，相当于是一个排队机制。使用Curator框架进行加锁和释放锁

来源：

https://segmentfault.com/a/1190000038330434