分布式锁-阿里云开发者社区

一、分布式锁的特点

在同一时间内只能被同一个线程执行
这把锁需要是一把可重入锁,避免死锁
这把锁最好是一把阻塞锁
获取锁和释放锁的性能要高
具备锁失效机制，防止死锁，无法清除锁

针对分布式锁的实现，目前比较常用的有以下几种方案：

二、基于数据库实现分布式锁

（一）、基于数据库insert和delete实现

创建一张锁表，通过操作该表中的数据来实现，当我们要锁住某个方法或资源时，就在表中增加一条记录，要释放锁的时候删除这条记录。

CREATETABLEIFNOTEXISTS'methodlock'('method_name'varchar(128)notnulldefault''comment'方法名称',primarykey('methodlock')) ENGINE=InnoDB DEFAULT CHARSET=utf8;

当我们要锁住某个方法时，执行以下SQL:

insertinto methodLock (method_name)values('method_name');

因为我们对method_name做了主键约束，这里如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就可以认为操作成功的那个线程获得了该方法的锁，就可以执行我们的业务逻辑。

当方法执行完毕之后，想要释放锁的话，需要执行以下Sql:

deletefrom methodLock where method_name='method_name';

这种简单的数据插入和删除数据的实现会出现以下几个问题：

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

（二）、基于数据库排他锁实现

借助数据库中自带的锁来实现分布式锁，还用刚刚创建的表，可以通过数据库的排他锁来实现分布式锁。基于MySql的InnoDB引擎，可以试用以下方法来实现加锁操作：

publicbooleanlock(){
connection.setAutoCommit(false);
while(true){
try{
result=select*frommethodlockwheremethod_name='methodname'forupdate;
if(result!=null) returntrue;
         }catch(Exceptione){
         }
   } 
returnfalse;
}

在查询语句后面增加for update ，数据库会在查询过程中给数据库表增加排他锁（InnoDB引擎在加锁的时候，只有通过索引进行检索的时候才会使用行级锁，否则会使用表级锁。我们这里肯定希望使用行级锁）当某条记录被加上排他锁之后，其他线程无法再在改航记录上增加排他锁。

我们可以任务获得排他锁的线程即是获得分布式锁的线程，获取到锁之后，可以执行业务逻辑，业务逻辑执行完之后再通过以下方法解锁：

publicvodiunlock(){
connection.commit();
}

这种方法可以有效的解决上面提到的无法释放锁和阻塞锁的问题。for update语句会在执行成功后立即返回，在执行失败时一直处于阻塞状态，直到成功。锁定之后服务宕机，无法释放？使用这种方式，服务宕机之后数据库会自己把锁释放掉。但是还是无法直接解决数据库单点和可重入问题。

还有一个问题需要注意就是我们要使用排他锁来进行分布式锁的lock，那么一个排他锁长时间不提交，就会占用数据库连接。一旦类似的连接变得多了，就可能把数据库连接池撑爆。

三、基于redis实现分布式锁

基于redis分布式缓存实现锁机制，采用的是redis的特性

StringlocakKey="LOCK";
Stringvalue=Thread.currentThread().getId().toString();
//加锁和设置过期时间放在一个事务中,防止加锁成功后，程序崩溃，对KEY没有设置上过期时间，导致死锁。try {
Booleanresult=stringRedisTemplate.opsForValue().setIfAbsent(lockKey, 
value, 10, TimeUnit.SECONDS);
if(!result){
returnfalse;   
            }
//业务逻辑    }catch (Exceptione){
e.printStackTrace();    
}finally {
//释放锁相应的锁,为防止误删锁，需要判断value是不是自己的value,value是自己线程的IDif(value.equals(stringRedisTemplate.opsForValue().get(lockKey))){
stringRedisTemplate.opsForValue().delete(lockKey);
    }
}
returntrue;
}

上述代码还是有问题，因为线程1在还没有执行完成的时候，此时锁已经到达过期时间，此时线程2则会加锁成功。例如线程一执行需要15秒，但是锁的时间只有10秒，线程一未执行完，锁超期时间到达，自动解锁，线程2自动加锁。针对此种情况可单独开辟一个线程，对锁进行续命，每3秒钟获取value值，如果value值和当前线程的值一直则进行自动续期。

使用redission框架实现redis分布式锁

@AutowiredRedissonredisson;
@RequestMapping("/deduct_stock_redisson")
publicStringdeductStockRedisson() {
StringlockKey="lock_key";
RLockrlock=redisson.getLock(lockKey);
try {
rlock.lock();
//业务逻辑实现，扣减库存        ....
    } catch (Exceptione) {
e.printStackTrace();
    } finally {
rlock.unlock();
    }
return"end";
}

多个线程去执行lock操作，仅有一个线程能够加锁成功，其它线程循环阻塞。
加锁成功，锁超时时间默认30s，并开启后台线程，加锁的后台会每隔10秒去检测线程持有的锁是否存在，还在的话，就延迟锁超时时间，重新设置为30s，即锁延期。
对于原子性，Redis分布式锁底层借助Lua脚本实现锁的原子性。锁延期是通过在底层用Lua进行延时，延时检测时间是对超时时间timeout /3

采用redis实现分布式锁，还会存在锁同步的问题，当主Redis加锁了，开始执行线程，若还未将锁通过异步同步的方式同步到从Redis节点，主节点就挂了，此时会把某一台从节点作为新的主节点，此时别的线程就可以加锁了，这样就出错了，怎么办？

使用zookeeper集群替代redis集群，实现同步。
使用红锁redlock算法，搭建多个redis,一般是奇数个，但是redis之间没有任何关系。

redis1

超过半数redis节点加锁成

功才算加锁成功

加锁

Java

redis2

加锁

Client

加锁

redis3

假设有3个redis节点，这些节点之间既没有主从，也没有集群关系。客户端用相同的key和随机值在3个节点上请求锁，请求锁的超时时间应小于锁自动释放时间。当在2个（超过半数）redis上请求到锁的时候，才算是真正获取到了锁。如果没有获取到锁，则把部分已锁的redis释放掉。

@RequestMapping("/deduct_stock_redlock")
publicStringdeductStockRedlock() {
StringlockKey="lock_key";
//TODO 这里需要自己实例化不同redis实例的redisson客户端连接，这里只是伪代码用一个redisson客户端简化了RLockrLock1=redisson.getLock(lockKey);
RLockrLock2=redisson.getLock(lockKey);
RLockrLock3=redisson.getLock(lockKey);
// 向3个redis实例尝试加锁RedissonRedLockredLock=newRedissonRedLock(rLock1, rLock2, rLock3);
booleanisLock;
try {
// 500ms拿不到锁, 就认为获取锁失败。10000ms即10s是锁失效时间。isLock=redLock.tryLock(500, 10000, TimeUnit.MILLISECONDS);
System.out.println("isLock = "+isLock);
if (isLock) {
//业务逻辑处理            ...
        }
    } catch (Exceptione) {
    } finally {
// 无论如何, 最后都要解锁redLock.unlock();
    }
}

如果采用以上redlok需要提高并发，可采用

四、基于zookeeper实现分布式锁

Zookeeper是基于临时有序节点实现分布式锁，当客户端需要加锁时候去zookeeper中创建一个目录，并生成一个瞬时有序节点，判断是否能获取锁的方式很简单，自己的节点是否是最小的一个。当使用完成后删除自己的临时节点，然后通知。

五、三种方案的比较

“没有银弹”，上面三种方案，无论那种方案都不可能做到完美。就像CAP一样，在复杂性、可靠性、性能等方面无法同时满足，所以，根据自己的需求选择适合自己的方案才是王道。

从理解的难易程度角度（从低到高）

数据库 > 缓存 > Zookeeper

从实现的复杂性角度（从低到高）

Zookeeper >= 缓存 > 数据库

从性能角度（从高到低）

缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）

Zookeeper > 缓存 > 数据库

分布式锁