分布式锁的原理和三种实用方式
我们经常提到的锁是在一个JVM内,加线程同步锁,例如 synchronize 或者 Lock锁,分布式的场景在多个服务之间,例如多个RPC接口之间,他们在不同的JVM内
例如多个RSF接口共同操作同一个资源的时候就要考虑到用一个公共资源来实现分布式锁的控制.
谈到分布式锁，有很多实现方式，如数据库、redis、ZooKeeper等。
1.1方案一(redis)
利用redis单线程的特性和setnx命令加锁，其中key是锁，value是锁的过期时间，
1. 通过setnx()方法尝试加锁，如果当前锁不存在，返回加锁成功。
2. 如果锁已经存在则获取锁的过期时间，和当前时间比较，如果锁已经过期，则设置新的过期时间，返回加锁成功。
代码如下:
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import com.alibaba.druid.util.StringUtils;
public class RedisLock {
@Autowired
private StringRedisTemplate redisTemplate;
/**
*
* 加锁: <br>
* redis实现的分布式锁
*
* @param key 锁的key
* @param value 当前时间+超时时间
* @return
* @see [相关类/方法](可选)
* @since [产品/模块版本](可选)
*/
public boolean lock(String key,String value){
//1.setIfAbsent 就是 对应redis 的 setnx
if(redisTemplate.opsForValue().setIfAbsent(key, value)){
//说明锁住了 就返回true
return true;
}
//先用get取出锁的有效时间
String currentValue = redisTemplate.opsForValue().get(key);
//如果 比对 系统当前时间 > currentValue,就说明 锁过期了
//就是 该过期时间, currentValue=A , 两个线程的value 多是B,其中一个线程拿到锁
if(!StringUtils.isEmpty(currentValue)
&& System.currentTimeMillis() > Long.parseLong(currentValue) ){
//获取上一个锁的时间
String oldValue = redisTemplate.opsForValue().getAndSet(key, value);
//最后再recheck一次, 如果有多个进来, 如果不相等,说明其它线程 也 获取到了锁
if(!StringUtils.isEmpty(oldValue) && oldValue.equals(currentValue)){
//获取到锁 返回true
return true;
}
}
//其它返回false
return false;
}
/**
*
* 功能描述: <br>
* 解锁
*
* @param key
* @param value
* @see [相关类/方法](可选)
* @since [产品/模块版本](可选)
*/
public void unlock(String key,String value){
try {
String currentValue = redisTemplate.opsForValue().get(key);
if(!StringUtils.isEmpty(currentValue) && currentValue.equals(value)){
redisTemplate.opsForValue().getOperations().delete(key);
}
} catch (Exception e) {
//做日志
}
}
public static void main(String[] args) {
RedisLock lock = new RedisLock();
long timeout = 60 * 5 * 1000l;//5分钟
long time = System.currentTimeMillis()+timeout ;
String productId = "1001";
lock.lock(productId,String.valueOf(time));
lock.unlock(productId, String.valueOf(time));
}
1.2方案二(基于zookeeper)
基于ZooKeeper实现分布式锁的步骤如下：
（1）创建一个目录mylock；
（2）线程A想获取锁就在mylock目录下创建临时顺序节点；
（3）获取mylock目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁；
（4）线程B获取所有节点，判断自己不是最小节点，设置监听比自己次小的节点；
（5）线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是不是最小的节点，如果是则获得锁。
这里推荐一个Apache的开源库Curator，它是一个ZooKeeper客户端，Curator提供的InterProcessMutex是分布式锁的实现，acquire方法用于获取锁，release方法用于释放锁。
acquire() :入参为空，调用该方法后，会一直堵塞，直到抢夺到锁资源，或者zookeeper连接中断后，上抛异常。
acquire(long time, TimeUnit unit)：入参传入超时时间以及单位，抢夺时，如果出现堵塞，会在超过该时间后，返回false。
对比两种方式，可以选择适合自己业务逻辑的方法。但是一般情况下，我推荐后者，传入超时时间，避免出现大量的临时节点累积以及线程堵塞的问题。
curator提供了四种锁, 包括读写锁,排他锁,
InterProcessMutex：分布式可重入排它锁
InterProcessSemaphoreMutex：分布式排它锁
InterProcessReadWriteLock：分布式读写锁
InterProcessMultiLock：将多个锁作为单个实体管理的容器
优点：具备高可用、可重入、阻塞锁特性，可解决失效死锁问题。
缺点：因为需要频繁的创建和删除节点，性能上不如Redis方式。
3.基于数据库的解决方法。Innodb支持行锁
基于数据库的实现方式的核心思想是：在数据库中创建一个表，表中包含方法名等字段，并在方法名字段上创建唯一索引，想要执行某个方法，就使用这个方法名向表中插入数据，成功插入则获取锁，执行完成后删除对应的行数据释放锁。
（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）想要执行某个方法，就使用这个方法名向表中插入数据：
INSERT INTO method_lock (method_name, desc) VALUES ('methodName', '测试的methodName');
因为我们对method_name做了唯一性约束，这里如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就可以认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。
（3）成功插入则获取锁，执行完成后删除对应的行数据释放锁：
delete from method_lock where method_name ='methodName';1
注意：这只是使用基于数据库的一种方法，使用数据库实现分布式锁还有很多其他的玩法！
总结:
哪种方式都无法做到完美。就像CAP一样，在复杂性、可靠性、性能等方面无法同时满足，所以，根据不同的应用场景选择最适合自己的才是王道。
从理解的难易程度角度（从低到高）
数据库 > 缓存 > Zookeeper
从实现的复杂性角度（从低到高）
Zookeeper >= 缓存 > 数据库
从性能角度（从高到低）
缓存 > Zookeeper >= 数据库
从可靠性角度（从高到低）
Zookeeper > 缓存 > 数据库
