分布式锁的三种实现
分布式锁的三种实现
在单体时代使用ReentrantLock、synchronized等来实锁，以便保证资源的安全性。但是在分布式时代，则需要分布式锁才能实现资源的安全性。
常见分布式的实现方式三种
- 数据库
rediszookeeper
本文咱们就针对上面三个分别来实现；
数据库实现分布式锁
首先，我们应该创建一个锁表，通过创建和查询数据来保证一个数据的原子性：
CREATE TABLE `order` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`order_no` int(11) DEFAULT NULL,
`pay_money` decimal(10, 2) DEFAULT NULL,
`status` int(4) DEFAULT NULL,
`create_date` datetime(0) DEFAULT NULL,
`delete_flag` int(4) DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE,
INDEX `idx_status`(`status`) USING BTREE,
INDEX `idx_order`(`order_no`) USING BTREE
) ENGINE = InnoDB其次，如果是校验订单的幂等性，就要先查询该记录是否存在数据库中，查询的时候要防止幻读，如果不存在，就插入到数据库，否则，放弃操作。
select id from `order` where `order_no`= 'xxxx' for update最后注意下，除了查询时防止幻读，我们还需要保证查询和插入是在同一个事务中，因此我们需要申明事务，具体的实现代码如下：
@Transactional
public int addOrderRecord(Order order) {
if(orderDao.selectOrderRecord(order)==null){
int result = orderDao.addOrderRecord(order);
if(result>0){
return 1;
}
}
return 0;
}到这，我们订单幂等性校验的分布式锁就实现了。我想你应该能发现为什么这种方式会存在性能瓶颈了。我们知道在 RR 事务级别，select 的 for update 操作是基于间隙锁 gap lock 实现的，这是一种悲观锁的实现方式，所以存在阻塞问题。
因此在高并发情况下，当有大量的请求进来时，大部分的请求都会进行排队等待。为了保证数据库的稳定性，事务的超时时间往往又设置得很小，所以就会出现大量事务被中断的情况。
除了阻塞等待之外，因为订单没有删除操作，所以这张锁表的数据将会逐渐累积，我们需要设置另外一个线程，隔一段时间就去删除该表中的过期订单，这就增加了业务的复杂度。
除了这种幂等性校验的分布式锁，有一些单纯基于数据库实现的分布式锁代码块或对象，是需要在锁释放时，删除或修改数据的。如果在获取锁之后，锁一直没有获得释放，即数据没有被删除或修改，这将会引发死锁问题。
基于 Redis 的分布式锁
利用 SETNX 和 SETEX
基本命令主要有：
- SETNX(SET If Not Exists)：当且仅当 Key 不存在时，则可以设置，否则不做任何动作。
- SETEX：可以设置超时时间
其原理为：通过 SETNX 设置 Key-Value 来获得锁，随即进入死循环，每次循环判断，如果存在 Key 则继续循环，如果不存在 Key，则跳出循环，当前任务执行完成后，删除 Key 以释放锁。
这种方式可能会导致死锁，为了避免这种情况，需要设置超时时间。
下面，请看具体的实现步骤。
1.创建一个 Maven 工程并在 pom.xml 加入以下依赖：
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.0.2.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- 开启web-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- redis-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
</dependencies>2.创建启动类 Application.java：
@SpringBootApplication
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class,args);
}
}3.添加配置文件 application.yml：
server:
port: 8080
spring:
redis:
host: localhost
port: 63794.创建全局锁类 Lock.java：
/**
* 全局锁，包括锁的名称
*/
public class Lock {
private String name;
private String value;
public Lock(String name, String value) {
this.name = name;
this.value = value;
}
public String getName() {
return name;
}
public String getValue() {
return value;
}
}5.创建分布式锁类 DistributedLockHandler.java：
@Component
public class DistributedLockHandler {
private static final Logger logger = LoggerFactory.getLogger(DistributedLockHandler.class);
private final static long LOCK_EXPIRE = 30 * 1000L;//单个业务持有锁的时间30s，防止死锁
private final static long LOCK_TRY_INTERVAL = 30L;//默认30ms尝试一次
private final static long LOCK_TRY_TIMEOUT = 20 * 1000L;//默认尝试20s
@Autowired
private StringRedisTemplate template;
/**
* 尝试获取全局锁
*
* @param lock 锁的名称
* @return true 获取成功，false获取失败
*/
public boolean tryLock(Lock lock) {
return getLock(lock, LOCK_TRY_TIMEOUT, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
}
/**
* 尝试获取全局锁
*
* @param lock 锁的名称
* @param timeout 获取超时时间 单位ms
* @return true 获取成功，false获取失败
*/
public boolean tryLock(Lock lock, long timeout) {
return getLock(lock, timeout, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
}
/**
* 尝试获取全局锁
*
* @param lock 锁的名称
* @param timeout 获取锁的超时时间
* @param tryInterval 多少毫秒尝试获取一次
* @return true 获取成功，false获取失败
*/
public boolean tryLock(Lock lock, long timeout, long tryInterval) {
return getLock(lock, timeout, tryInterval, LOCK_EXPIRE);
}
/**
* 尝试获取全局锁
*
* @param lock 锁的名称
* @param timeout 获取锁的超时时间
* @param tryInterval 多少毫秒尝试获取一次
* @param lockExpireTime 锁的过期
* @return true 获取成功，false获取失败
*/
public boolean tryLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
return getLock(lock, timeout, tryInterval, lockExpireTime);
}
/**
* 操作redis获取全局锁
*
* @param lock 锁的名称
* @param timeout 获取的超时时间
* @param tryInterval 多少ms尝试一次
* @param lockExpireTime 获取成功后锁的过期时间
* @return true 获取成功，false获取失败
*/
public boolean getLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
try {
if (StringUtils.isEmpty(lock.getName()) || StringUtils.isEmpty(lock.getValue())) {
return false;
}
long startTime = System.currentTimeMillis();
do{
if (!template.hasKey(lock.getName())) {
ValueOperations<String, String> ops = template.opsForValue();
ops.set(lock.getName(), lock.getValue(), lockExpireTime, TimeUnit.MILLISECONDS);
return true;
} else {//存在锁
logger.debug("lock is exist!！！");
}
if (System.currentTimeMillis() - startTime > timeout) {//尝试超过了设定值之后直接跳出循环
return false;
}
Thread.sleep(tryInterval);
}
while (template.hasKey(lock.getName())) ;
} catch (InterruptedException e) {
logger.error(e.getMessage());
return false;
}
return false;
}
/**
* 释放锁
*/
public void releaseLock(Lock lock) {
if (!StringUtils.isEmpty(lock.getName())) {
template.delete(lock.getName());
}
}
}6.最后创建 HelloController 来测试分布式锁。
@RestController
public class HelloController {
@Autowired
private DistributedLockHandler distributedLockHandler;
@RequestMapping("index")
public String index(){
Lock lock=new Lock("lynn","min");
if(distributedLockHandler.tryLock(lock)){
try {
//为了演示锁的效果，这里睡眠5000毫秒
System.out.println("执行方法");
Thread.sleep(5000);
}catch (Exception e){
e.printStackTrace();
}
distributedLockHandler.releaseLock(lock);
}
return "hello world!";
}
}7.测试。
启动 Application.java，连续访问两次浏览器：http://localhost:8080/index，控制台可以发现先打印了一次“执行方法”，说明后面一个线程被锁住了，5秒后又再次打印了“执行方法”，说明锁被成功释放。
通过这种方式创建的分布式锁存在以下问题：
- 高并发的情况下，如果两个线程同时进入循环，可能导致加锁失败。
- SETNX 是一个耗时操作，因为它需要判断 Key 是否存在，因为会存在性能问题。
因此，Redis 官方推荐 Redlock 来实现分布式锁。
利用 Redlock
通过 Redlock 实现分布式锁比其他算法更加可靠，继续改造上一例的代码。
1.pom.xml 增加以下依赖：
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.7.0</version>
</dependency>2.增加以下几个类：
/**
* 获取锁后需要处理的逻辑
*/
public interface AquiredLockWorker<T> {
T invokeAfterLockAquire() throws Exception;
}/**
* 获取锁管理类
*/
public interface DistributedLocker {
/**
* 获取锁
* @param resourceName 锁的名称
* @param worker 获取锁后的处理类
* @param <T>
* @return 处理完具体的业务逻辑要返回的数据
* @throws UnableToAquireLockException
* @throws Exception
*/
<T> T lock(String resourceName, AquiredLockWorker<T> worker) throws UnableToAquireLockException, Exception;
<T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception;
}/**
* 异常类
*/
public class UnableToAquireLockException extends RuntimeException {
public UnableToAquireLockException() {
}
public UnableToAquireLockException(String message) {
super(message);
}
public UnableToAquireLockException(String message, Throwable cause) {
super(message, cause);
}
}/**
* 获取RedissonClient连接类
*/
@Component
public class RedissonConnector {
RedissonClient redisson;
@PostConstruct
public void init(){
redisson = Redisson.create();
}
public RedissonClient getClient(){
return redisson;
}
}@Component
public class RedisLocker implements DistributedLocker{
private final static String LOCKER_PREFIX = "lock:";
@Autowired
RedissonConnector redissonConnector;
@Override
public <T> T lock(String resourceName, AquiredLockWorker<T> worker) throws InterruptedException, UnableToAquireLockException, Exception {
return lock(resourceName, worker, 100);
}
@Override
public <T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception {
RedissonClient redisson= redissonConnector.getClient();
RLock lock = redisson.getLock(LOCKER_PREFIX + resourceName);
// Wait for 100 seconds seconds and automatically unlock it after lockTime seconds
boolean success = lock.tryLock(100, lockTime, TimeUnit.SECONDS);
if (success) {
try {
return worker.invokeAfterLockAquire();
} finally {
lock.unlock();
}
}
throw new UnableToAquireLockException();
}
}3.修改 HelloController：
@RestController
public class HelloController {
@Autowired
private DistributedLocker distributedLocker;
@RequestMapping("index")
public String index()throws Exception{
distributedLocker.lock("test",new AquiredLockWorker<Object>() {
@Override
public Object invokeAfterLockAquire() {
try {
System.out.println("执行方法！");
Thread.sleep(5000);
}catch (Exception e){
e.printStackTrace();
}
return null;
}
});
return "hello world!";
}
}4.按照上节的测试方法进行测试，我们发现分布式锁也生效了。
Redlock 是 Redis 官方推荐的一种方案，因此可靠性比较高。
Zookeeper 实现分布式锁
除了数据库实现分布式锁的方式以外，我们还可以基于 Zookeeper 实现。Zookeeper 是一种提供“分布式服务协调“的中心化服务，正是 Zookeeper 的以下两个特性，分布式应用程序才可以基于它实现分布式锁功能。
顺序临时节点：Zookeeper 提供一个多层级的节点命名空间（节点称为 Znode），每个节点都用一个以斜杠（/）分隔的路径来表示，而且每个节点都有父节点（根节点除外），非常类似于文件系统。
节点类型可以分为持久节点（PERSISTENT ）、临时节点（EPHEMERAL），每个节点还能被标记为有序性（SEQUENTIAL），一旦节点被标记为有序性，那么整个节点就具有顺序自增的特点。一般我们可以组合这几类节点来创建我们所需要的节点，例如，创建一个持久节点作为父节点，在父节点下面创建临时节点，并标记该临时节点为有序性。
Watch 机制：Zookeeper 还提供了另外一个重要的特性，Watcher（事件监听器）。ZooKeeper 允许用户在指定节点上注册一些 Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知给用户。
我们熟悉了 Zookeeper 的这两个特性之后，就可以看看 Zookeeper 是如何实现分布式锁的了。
首先，我们需要建立一个父节点，节点类型为持久节点（PERSISTENT） ，每当需要访问共享资源时，就会在父节点下建立相应的顺序子节点，节点类型为临时节点（EPHEMERAL），且标记为有序性（SEQUENTIAL），并且以临时节点名称 + 父节点名称 + 顺序号组成特定的名字。
在建立子节点后，对父节点下面的所有以临时节点名称 name 开头的子节点进行排序，判断刚刚建立的子节点顺序号是否是最小的节点，如果是最小节点，则获得锁。
如果不是最小节点，则阻塞等待锁，并且获得该节点的上一顺序节点，为其注册监听事件，等待节点对应的操作获得锁。
当调用完共享资源后，删除该节点，关闭 zk，进而可以触发监听事件，释放该锁。
以上实现的分布式锁是严格按照顺序访问的并发锁。一般我们还可以直接引用 Curator 框架来实现 Zookeeper 分布式锁，代码如下：
InterProcessMutex lock = new InterProcessMutex(client, lockPath);
if ( lock.acquire(maxWait, waitUnit) ){
try {
// do some work inside of the critical section here
}finally{
lock.release();
}
}Zookeeper 实现的分布式锁，例如相对数据库实现，有很多优点。Zookeeper 是集群实现，可以避免单点问题，且能保证每次操作都可以有效地释放锁，这是因为一旦应用服务挂掉了，临时节点会因为 session 连接断开而自动删除掉。
由于频繁地创建和删除结点，加上大量的 Watch 事件，对 Zookeeper 集群来说，压力非常大。且从性能上来说，其与接下来我要讲的 Redis 实现的分布式锁相比，还是存在一定的差距。