概述

为了防止分布式系统中的多个进程之间相互干扰，我们需要一种分布式协调技术来对这些进程进行调度。而这个分布式协调技术的核心就是来实现这个分布式锁。

为什么要使用分布式锁

file

成员变量 A 存在 JVM1、JVM2、JVM3 三个 JVM 内存中
成员变量 A 同时都会在 JVM 分配一块内存，三个请求发过来同时对这个变量操作，显然结果是不对的
不是同时发过来，三个请求分别操作三个不同 JVM 内存区域的数据，变量 A 之间不存在共享，也不具有可见性，处理的结果也是不对的
注：该成员变量 A 是一个有状态的对象
如果我们业务中确实存在这个场景的话，我们就需要一种方法解决这个问题，这就是分布式锁要解决的问题

分布式锁应该具有哪些特点

在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；
高可用的获取锁与释放锁；
高性能的获取锁与释放锁；
具备可重入特性；
具备锁失效机制，防止死锁；
具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败。

实现分布式锁的N种方式

数据库实现排他锁
Redis：和 Memcached 的方式类似，利用 Redis 的 setnx 命令。此命令同样是原子性操作，只有在 key 不存在的情况下，才能 set 成功。
Zookeeper：利用 Zookeeper 的顺序临时节点，来实现分布式锁和等待队列。Zookeeper 设计的初衷，就是为了实现分布式锁服务的。
Memcached：利用 Memcached 的 add 命令。此命令是原子性操作，只有在 key 不存在的情况下，才能 add 成功，也就意味着线程得到了锁。
Chubby：Google 公司实现的粗粒度分布式锁服务，底层利用了 Paxos 一致性算法。
接下来，介绍三种常用的方式

1.基于Mysql实现分布式锁

表结构

-- auto-generated definition
create table distributed_lock
(id          int auto_incrementprimary key,method_name varchar(100)                       not null comment '获取锁的方法名',remark      varchar(100)                       null comment '备注信息',status      int                                not null comment '分配状态: 1-未分配，2-已分配',update_time datetime default CURRENT_TIMESTAMP not null on update CURRENT_TIMESTAMP,version     int                                not null comment '版本号',constraint uidx_method_nameunique (method_name)
)comment '分布式锁表';

先获取锁的信息

select id, method_name, status,version from distributed_lock where status=1 and method_name='methodName';```### 占有锁

update t_resoure set status=2, version=2, update_time=now() where method_name=‘methodName’ and status=1 and version=2;```

如果没有更新影响到一行数据，则说明这个资源已经被别人占位了。
以上也是cap理论的悲观锁，在数据库上面的应用。

缺点：

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

解决方案：

数据库是单点？搞两个数据库，数据之前双向同步。一旦挂掉快速切换到备库上。
没有失效时间？只要做一个定时任务，每隔一定时间把数据库中的超时数据清理一遍。
非阻塞的？搞一个while循环，直到insert成功再返回成功。
非重入的？在数据库表中加个字段，记录当前获得锁的机器的主机信息和线程信息，那么下次再获取锁的时候先查询数据库，如果当前机器的主机信息和线程信息在数据库可以查到的话，直接把锁分配给他就可以了。

2.基于Redis实现分布式锁

file

首先要弄清楚几个redis命令的概念

setnx()

setnx 的含义就是 SET if Not Exists，其主要有两个参数 setnx(key, value)。该方法是原子的，如果 key 不存在，则设置当前 key 成功，返回 1；如果当前 key 已经存在，则设置当前 key 失败，返回 0。

expire()

expire 设置过期时间，要注意的是 setnx 命令不能设置 key 的超时时间，只能通过 expire() 来对 key 设置。

getset()

这个命令主要有两个参数 getset(key，newValue)。该方法是原子的，对 key 设置 newValue 这个值，并且返回 key 原来的旧值。假设 key 原来是不存在的，那么多次执行这个命令，会出现下边的效果：
getset(key, “value1”) 返回 null 此时 key 的值会被设置为 value1
getset(key, “value2”) 返回 value1 此时 key 的值会被设置为 value2
依次类推！

使用步骤：

1.setnx(lockkey, 当前时间过期超时时间)，如果返回 1，则获取锁成功；如果返回 0 则没有获取到锁，转向 2。

2.get(lockkey) 获取值 oldExpireTime ，并将这个 value 值与当前的系统时间进行比较，如果小于当前系统时间，则认为这个锁已经超时，可以允许别的请求重新获取，转向 3。

3.计算 newExpireTime = 当前时间过期超时时间，然后 getset(lockkey, newExpireTime) 会返回当前 lockkey 的值currentExpireTime。

4.判断 currentExpireTime 与 oldExpireTime 是否相等，如果相等，说明当前 getset 设置成功，获取到了锁。如果不相等，说明这个锁又被别的请求获取走了，那么当前请求可以直接返回失败，或者继续重试。

5.在获取到锁之后，当前线程可以开始自己的业务处理，当处理完毕后，比较自己的处理时间和对于锁设置的超时时间，如果小于锁设置的超时时间，则直接执行 delete 释放锁；如果大于锁设置的超时时间，则不需要再锁进行处理。

话不多说，上代码:

public final class RedisLockUtil {
private static final int defaultExpire = 60;
private RedisLockUtil() {//}
/*** 加锁* @param key redis key* @param expire 过期时间，单位秒* @return true:加锁成功，false，加锁失败*/public static boolean lock(String key, int expire) {
RedisService redisService = SpringUtils.getBean(RedisService.class);long status = redisService.setnx(key, "1");
if(status == 1) {redisService.expire(key, expire);return true;}
return false;}
public static boolean lock(String key) {return lock2(key, defaultExpire);}
/*** 加锁* @param key redis key* @param expire 过期时间，单位秒* @return true:加锁成功，false，加锁失败*/public static boolean lock2(String key, int expire) {
RedisService redisService = SpringUtils.getBean(RedisService.class);
long value = System.currentTimeMillis()   expire;long status = redisService.setnx(key, String.valueOf(value));
if(status == 1) {return true;}long oldExpireTime = Long.parseLong(redisService.get(key, "0"));if(oldExpireTime < System.currentTimeMillis()) {//超时long newExpireTime = System.currentTimeMillis()   expire;long currentExpireTime = Long.parseLong(redisService.getSet(key, String.valueOf(newExpireTime)));if(currentExpireTime == oldExpireTime) {return true;}}return false;}
public static void unLock1(String key) {RedisService redisService = SpringUtils.getBean(RedisService.class);redisService.del(key);}
public static void unLock2(String key) {    RedisService redisService = SpringUtils.getBean(RedisService.class);    long oldExpireTime = Long.parseLong(redisService.get(key, "0"));   if(oldExpireTime > System.currentTimeMillis()) {        redisService.del(key);    }}
}

3.基于Zookeeper实现分布式锁

首先，我们先来看看zookeeper的相关知识。

zk 一般由多个节点构成（单数），采用 zab 一致性协议。因此可以将 zk 看成一个单点结构，对其修改数据其内部自动将所有节点数据进行修改而后才提供查询服务。

zk 的数据以目录树的形式，每个目录称为 znode， znode 中可存储数据（一般不超过 1M），还可以在其中增加子节点。
子节点有三种类型。序列化节点，每在该节点下增加一个节点自动给该节点的名称上自增。临时节点，一旦创建这个 znode 的客户端与服务器失去联系，这个 znode 也将自动删除。最后就是普通节点。
Watch 机制，client 可以监控每个节点的变化，当产生变化会给 client 产生一个事件。

zookeeper基本锁：

**原理：**利用临时节点与 watch 机制。每个锁占用一个普通节点 /lock，当需要获取锁时在 /lock 目录下创建一个临时节点，创建成功则表示获取锁成功，失败则 watch/lock 节点，有删除操作后再去争锁。临时节点好处在于当进程挂掉后能自动上锁的节点自动删除即取消锁。

**缺点：**所有取锁失败的进程都监听父节点，很容易发生羊群效应，即当释放锁后所有等待进程一起来创建节点，并发量很大。

zookeeper分布式锁实现:

**原理：**上锁改为创建临时有序节点，每个上锁的节点均能创建节点成功，只是其序号不同。只有序号最小的可以拥有锁，如果这个节点序号不是最小的则 watch 序号比本身小的前一个节点 (公平锁)。

步骤：

在 /lock 节点下创建一个有序临时节点 (EPHEMERAL_SEQUENTIAL)。
判断创建的节点序号是否最小，如果是最小则获取锁成功。不是则取锁失败，然后 watch 序号比本身小的前一个节点。
当取锁失败，设置 watch 后则等待 watch 事件到来后，再次判断是否序号最小。
取锁成功则执行代码，最后释放锁（删除该节点）。

代码如下:

public final class RedisLockUtil {
private static final int defaultExpire = 60;
private RedisLockUtil() {//}
/*** 加锁* @param key redis key* @param expire 过期时间，单位秒* @return true:加锁成功，false，加锁失败*/public static boolean lock(String key, int expire) {
RedisService redisService = SpringUtils.getBean(RedisService.class);long status = redisService.setnx(key, "1");
if(status == 1) {redisService.expire(key, expire);return true;}
return false;}
public static boolean lock(String key) {return lock2(key, defaultExpire);}
/*** 加锁* @param key redis key* @param expire 过期时间，单位秒* @return true:加锁成功，false，加锁失败*/public static boolean lock2(String key, int expire) {
RedisService redisService = SpringUtils.getBean(RedisService.class);
long value = System.currentTimeMillis()   expire;long status = redisService.setnx(key, String.valueOf(value));
if(status == 1) {return true;}long oldExpireTime = Long.parseLong(redisService.get(key, "0"));if(oldExpireTime < System.currentTimeMillis()) {//超时long newExpireTime = System.currentTimeMillis()   expire;long currentExpireTime = Long.parseLong(redisService.getSet(key, String.valueOf(newExpireTime)));if(currentExpireTime == oldExpireTime) {return true;}}return false;}
public static void unLock1(String key) {RedisService redisService = SpringUtils.getBean(RedisService.class);redisService.del(key);}
public static void unLock2(String key) {    RedisService redisService = SpringUtils.getBean(RedisService.class);    long oldExpireTime = Long.parseLong(redisService.get(key, "0"));   if(oldExpireTime > System.currentTimeMillis()) {        redisService.del(key);    }}
}