分布式锁
1、什么是分布式锁？
当多个进程在同一个系统中，用分布式锁控制多个进程对资源的访问。传统的单体应用单机部署情况下，可以使用java并发处理相关的API进行互斥控制。分布式系统后由于多线程，多进程分布在不同机器上，使单机部署情况下的并发控制锁策略失效，为了解决跨JVM互斥机制来控制共享资源的访问，这就是分布式锁的来源；分布式锁应用场景大都是高并发、大流量场景。
2、分布式锁实现
（1）、redis分布式锁的实现
加锁机制：根据hash节点选择一个客户端执行lua脚本
锁互斥机制：再来一个客户端执行同样的lua脚本会提示已经存在锁，然后进入循环一直尝试加锁
可重入机制
watch dog自动延期机制
释放锁机制
测试用例
单机
private RedissonClient getClient(){
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");//.setPassword("");//.setConnectionMinimumIdleSize(10).setConnectionPoolSize(10);//.setConnectionPoolSize();//172.16.10.164
RedissonClient redissonClient = Redisson.create(config);
return redissonClient;
}
private ExecutorService executorService = Executors.newCachedThreadPool();
@Test
public void test() throws Exception {
int[] count = {0};
for (int i = 0; i < 10; i++) {
RedissonClient client = getClient();
final RedisLock redisLock = new RedisLock(client,"lock_key");
executorService.submit(() -> {
try {
redisLock.lock();
count[0]++;
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
redisLock.unlock();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.HOURS);
System.out.println(count[0]);
}
RedLock
public static RLock create (String url, String key){
Config config = new Config();
config.useSingleServer().setAddress(url);
RedissonClient redissonClient = Redisson.create(config);
return redissonClient.getLock(key);
}
RedissonRedLock redissonRedLock = new RedissonRedLock(
create("redis://redis://127.0.0.1:6379","lock_key1"),
create("redis://redis://127.0.0.1:6380","lock_key2"),
create("redis://redis://127.0.0.1:6381","lock_key3"));
RedisRedLock redLock = new RedisRedLock(redissonRedLock);
private ExecutorService executorService = Executors.newCachedThreadPool();
@Test
public void test() throws Exception {
int[] count = {0};
for (int i = 0; i < 2; i++) {
executorService.submit(() -> {
try {
redLock.lock();
count[0]++;
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
redLock.unlock();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.HOURS);
System.out.println(count[0]);
}
利用StringRedisTemplate，分布式锁工具类
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.util.concurrent.TimeUnit;
@Component
public class RedisLock {
private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 加锁
*
* @param key - 唯一标志
* @param value 跟唯一标志对应的随机值
* @return
*/
public boolean lock(String key, String value) {
//设置30秒的锁
return stringRedisTemplate.opsForValue().setIfAbsent(key, value, 30, TimeUnit.SECONDS);//对应setnx命令
}
/**
* 解锁
*
* @param key
* @param value
*/
public void unlock(String key, String value) {
try {
String currentValue = stringRedisTemplate.opsForValue().get(key);
if (!StringUtils.isEmpty(currentValue) && currentValue.equals(value)) {
stringRedisTemplate.opsForValue().getOperations().delete(key);//删除key
}
} catch (Exception e) {
logger.error("[Redis分布式锁] 解锁出现异常", e);
}
}
}
String uuid = UUID.randomUUID().toString();
boolean lock = redisLock.lock("lock_" + id, uuid);
if (lock) {
// xxxxxx
}
redisLock.unlock("lock_" + id, uuid);
（2）、基于ETCD实现分布式锁分析
ETCD分布式锁的实现
Lease机制：租约机制（TTL，Time To Live），Etcd 可以为存储的 key-value 对设置租约，当租约到期，key-value 将失效删除；同时也支持续约，通过客户端可以在租约到期之前续约，以避免 key-value 对过期失效。Lease 机制可以保证分布式锁的安全性，为锁对应的 key 配置租约，即使锁的持有者因故障而不能主动释放锁，锁也会因租约到期而自动释放。
Revision机制：每个 key 带有一个 Revision 号，每进行一次事务加一，它是全局唯一的，通过 Revision 的大小就可以知道进行写操作的顺序。在实现分布式锁时，多个客户端同时抢锁，根据 Revision 号大小依次获得锁，可以避免 “羊群效应” ，实现公平锁。
Prefix机制：即前缀机制。例如，一个名为 /etcdlock 的锁，两个争抢它的客户端进行写操作，实际写入的 key 分别为：key1="/etcdlock/UUID1"，key2="/etcdlock/UUID2"，其中，UUID 表示全局唯一的 ID，确保两个 key 的唯一性。写操作都会成功，但返回的 Revision 不一样，那么，如何判断谁获得了锁呢？通过前缀 /etcdlock 查询，返回包含两个 key-value 对的的 KeyValue 列表，同时也包含它们的 Revision，通过 Revision 大小，客户端可以判断自己是否获得锁。
Watch机制：即监听机制，Watch 机制支持 Watch 某个固定的 key，也支持 Watch 一个范围（前缀机制），当被 Watch 的 key 或范围发生变化，客户端将收到通知；在实现分布式锁时，如果抢锁失败，可通过 Prefix 机制返回的 KeyValue 列表获得 Revision 比自己小且相差最小的 key（称为 pre-key），对 pre-key 进行监听，因为只有它释放锁，自己才能获得锁，如果 Watch 到 pre-key 的 DELETE 事件则说明 pre-key 已经释放，自己已经持有锁。
基于ETCD分布式锁
**步骤1：**建立连接
客户端连接 Etcd，以 /etcd/lock 为前缀创建全局唯一的 key，假设第一个客户端对应的 key="/etcd/lock/UUID1"，第二个为 key="/etcd/lock/UUID2"；客户端分别为自己的 key 创建租约 - Lease，租约的长度根据业务耗时确定；
**步骤2：**创建定时任务作为租约的“心跳”
当一个客户端持有锁期间，其它客户端只能等待，为了避免等待期间租约失效，客户端需创建一个定时任务作为“心跳”进行续约。此外，如果持有锁期间客户端崩溃，心跳停止，key 将因租约到期而被删除，从而锁释放，避免死锁。
**步骤3：**客户端将自己全局唯一的 key 写入 Etcd
执行 put 操作，将步骤 1 中创建的 key 绑定租约写入 Etcd，根据 Etcd 的 Revision 机制，假设两个客户端 put 操作返回的 Revision 分别为 1、2，客户端需记录 Revision 用以接下来判断自己是否获得锁
**步骤 4：**客户端判断是否获得锁
客户端以前缀 /etcd/lock/ 读取 keyValue 列表，判断自己 key 的 Revision 是否为当前列表中最小的，如果是则认为获得锁；否则监听列表中前一个 Revision 比自己小的 key 的删除事件，一旦监听到删除事件或者因租约失效而删除的事件，则自己获得锁。
**步骤 5：**执行业务
获得锁后，操作共享资源，执行业务代码
**步骤 6：**释放锁
完成业务流程后，删除对应的key释放锁
例子：
public class EtcdDistributeLock extends AbstractLock{
private Client client;
private Lock lockClient;
private Lease leaseClient;
private String lockKey;
private String lockPath;
/** 锁的次数 */
private AtomicInteger lockCount;
/** 租约有效期,防止客户端崩溃，可在租约到期后自动释放锁；另一方面，正常执行过程中，会自动进行续租,单位 ns */
private Long leaseTTL;
/** 续约锁租期的定时任务，初次启动延迟，单位默认为 s,默认为1s，可根据业务定制设置*/
private Long initialDelay = 0L;
/** 定时任务线程池类 */
ScheduledExecutorService service = null;
/** 保存线程与锁对象的映射，锁对象包含重入次数，重入次数的最大限制为Int的最大值 */
private final ConcurrentMap<Thread, LockData> threadData = Maps.newConcurrentMap();
public EtcdDistributeLock(){}
public EtcdDistributeLock(Client client, String lockKey, long leaseTTL,TimeUnit unit){
this.client = client;
lockClient = client.getLockClient();
leaseClient = client.getLeaseClient();
this.lockKey = lockKey;
// 转纳秒
this.leaseTTL = unit.toNanos(leaseTTL);
service = Executors.newSingleThreadScheduledExecutor();
}
@Override
public void lock() {
// 检查重入性
Thread currentThread = Thread.currentThread();
LockData oldLockData = threadData.get(currentThread);
if (oldLockData != null && oldLockData.isLockSuccess()) {
// re-entering
int lockCount = oldLockData.lockCount.incrementAndGet();
if(lockCount < 0 ){
throw new Error("超出可重入次数限制");
}
return;
}
// 记录租约 ID
Long leaseId = 0L;
try{
leaseId = leaseClient.grant(TimeUnit.NANOSECONDS.toSeconds(leaseTTL)).get().getID();
// 续租心跳周期
long period = leaseTTL - leaseTTL / 5;
// 启动定时任务续约
service.scheduleAtFixedRate(new EtcdDistributeLock.KeepAliveRunnable(leaseClient, leaseId),
initialDelay,period,TimeUnit.NANOSECONDS);
LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get();
if(lockResponse != null){
lockPath = lockResponse.getKey().toString(Charset.forName("utf-8"));
log.info("获取锁成功,锁路径:{},线程:{}",lockPath,currentThread.getName());
}
}catch (InterruptedException | ExecutionException e){
log.error("获取锁失败",e);
return;
}
// 获取锁成功，锁对象设置
LockData newLockData = new LockData(currentThread, lockKey);
newLockData.setLeaseId(leaseId);
newLockData.setService(service);
threadData.put(currentThread, newLockData);
newLockData.setLockSuccess(true);
}
@Override
public void lockInterruptibly() throws InterruptedException {
super.lockInterruptibly();
}
@Override
public boolean tryLock() {
return super.tryLock();
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return super.tryLock(time,unit);
}
@Override
public void unlock() {
Thread currentThread = Thread.currentThread();
LockData lockData = threadData.get(currentThread);
if (lockData == null){
throw new IllegalMonitorStateException("You do not own the lock: " + lockKey);
}
int newLockCount = lockData.lockCount.decrementAndGet();
if ( newLockCount > 0 ) {
return;
}
if ( newLockCount < 0 ) {
throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + lockKey);
}
try {
// 释放锁
if(lockPath != null){
lockClient.unlock(ByteSequence.from(lockPath.getBytes())).get();
}
if(lockData != null){
// 关闭定时任务
lockData.getService().shutdown();
// 删除租约
if (lockData.getLeaseId() != 0L) {
leaseClient.revoke(lockData.getLeaseId());
}
}
} catch (InterruptedException | ExecutionException e) {
log.error("解锁失败",e);
}finally {
// 移除当前线程资源
threadData.remove(currentThread);
}
}
@Override
public Condition newCondition() {
return super.newCondition();
}
/**
* 心跳续约线程类
*/
public static class KeepAliveRunnable implements Runnable {
private Lease leaseClient;
private long leaseId;
public KeepAliveRunnable(Lease leaseClient, long leaseId) {
this.leaseClient = leaseClient;
this.leaseId = leaseId;
}
@Override
public void run() {
// 对该leaseid进行一次续约
leaseClient.keepAliveOnce(leaseId);
}
}
public class EtcdLockTest {
private Client client;
private String key = "/etcd/lock";
private static final String server = "http://xxxx:xxxx";
private ExecutorService executorService = Executors.newFixedThreadPool(10000);
@Before
public void before() throws Exception {
initEtcdClient();
}
private void initEtcdClient(){
client = Client.builder().endpoints(server).build();
}
@Test
public void testEtcdDistributeLock() throws InterruptedException {
int[] count = {0};
for (int i = 0; i < 100; i++) {
executorService.submit(() -> {
final EtcdDistributeLock lock = new EtcdDistributeLock(client, key,20,TimeUnit.SECONDS);
try {
lock.lock();
count[0]++;
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
lock.unlock();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.HOURS);
System.err.println("执行结果: " + count[0]);
}
}
（3）、基于Zookeeper分布式锁
实现原理
启动客户端，确认链接到了服务器
多个客户端并发的在特定路径下创建临时性顺序节点
客户端判断自己的创建的顺序节点是否是最小的，如果是最小的，则获取锁成功
第三步若判定失败，则采用zk的watch机制监听自己的前一个顺序节点，等待前一个节点的删除（放锁）事件，再开始第三步判定
zookeeper作为高性能分布式协调框架，可以把其看做一个文件系统，其中有节点的概念，并且分为4种：1.持久性节点2.持久性顺序节点3.临时性节点4.临时性顺序节点。
分布式锁的实现主要思路就是：监控其他客户端的状态，来判断自己是否可以获得锁。
采用临时性顺序节点的原因：
zk服务器维护了客户端的会话有效性，当会话失效的时候，其会话所创建的临时性节点都会被删除，通过这一特点，可以通过watch临时节点来监控其他客户端的情况，方便自己做出相应动作。
因为zk对写操作是顺序性的，所以并发创建的顺序节点会有一个唯一确定的序号，当前锁是公平锁的一种实现，所以依靠这种顺序性可以很好的解释—节点序列小的获取到锁并且可以采用watch自己的前一个节点来避免惊群现象（这样watch事件的传播是线性的）。
例子：
public class ZKLock extends AbstractLock {
/**
* 1.Connect to zk
*/
private CuratorFramework client;
private InterProcessLock lock ;
public ZKLock(String zkAddress,String lockPath) {
// 1.Connect to zk
client = CuratorFrameworkFactory.newClient(
zkAddress,
new RetryNTimes(5, 5000)
);
client.start();
if(client.getState() == CuratorFrameworkState.STARTED){
log.info("zk client start successfully!");
log.info("zkAddress:{},lockPath:{}",zkAddress,lockPath);
}else{
throw new RuntimeException("客户端启动失败。。。");
}
this.lock = defaultLock(lockPath);
}
private InterProcessLock defaultLock(String lockPath ){
return new InterProcessMutex(client, lockPath);
}
@Override
public void lock() {
try {
this.lock.acquire();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@Override
public boolean tryLock() {
boolean flag ;
try {
flag=this.lock.acquire(0,TimeUnit.SECONDS);
} catch (Exception e) {
throw new RuntimeException(e);
}
return flag;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
boolean flag ;
try {
flag=this.lock.acquire(time,unit);
} catch (Exception e) {
throw new RuntimeException(e);
}
return flag;
}
@Override
public void unlock() {
try {
this.lock.release();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
private ExecutorService executorService = Executors.newCachedThreadPool();
@Test
public void testLock() throws Exception{
ZKLock zkLock = new ZKLock("xxxx:xxxx","/lockPath");
int[] num = {0};
long start = System.currentTimeMillis();
for(int i=0;i<200;i++){
executorService.submit(()->{
try {
zkLock.lock();
num[0]++;
} catch (Exception e){
throw new RuntimeException(e);
} finally {
zkLock.unlock();
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.HOURS);
log.info("耗时:{}",System.currentTimeMillis()-start);
System.out.println(num[0]);
}
3、总结
redis的分布式锁中redisson一般为单实例，当单实例不可用时，会阻塞业务流程。主从方式、主从数据异步，会存在锁失效的问题。RedLock一般要求至少3台以上的redis主从实例，维护成本相对来说比较高。
ZK锁具备高可用、可重入、阻塞锁特性，可解决失效死锁问题。但是因为需要频繁的创建和删除节点，性能上不如Redis方式。
ETCD分布式锁的实现原理与zk锁类似，但是ETCD分布式锁更加可靠强大。其Lease功能保证分布式锁的安全性；watch功能支持监听某个固定的key，也支持watch一个范围的key（前缀机制）；revision功能可通过 Revision 的大小就可以知道进行写操作的顺序。可以避免 “羊群效应” （也称 “惊群效应”），实现公平锁。前缀机制与watch功能配合使用解决了死锁问题。总之ETCD的灵感来源于Zookeeper,但实现的时候做了很多的改进，如：高负载下的稳定读写、数据模型的多版本并发控制、稳定的watch功能，通知订阅者监听值得变化、可以容忍脑裂现场的发生、客户端的协议使用gRPC协议,支持go、c++、java等。
4、以Redis为例子常用锁开发步骤
1）公共方法的提取
我们这里先定义一个 RedisLock 接口，代码如下所示：
public interface RedisLock {
/**
* 尝试加锁
*/
boolean tryLock(String key, long timeout, TimeUnit unit);
/**
* 解锁操作
*/
void releaseLock(String key);
}
2）实现
接下来，我们基于上面已经实现的分布式锁的思路，来实现这个接口，代码如果所示：
public class RedisLockImpl implements RedisLock {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Override
public boolean tryLock(String key, long timeout, TimeUnit unit) {
return stringRedisTemplate.opsForValue().setIfAbsent(key, "lock-test", timeout, unit);
}
@Override
public void releaseLock(String key) {
stringRedisTemplate.delete(key);
}
}
3）加锁&解锁的归一化
我们先来继续分析上面代码。从开发的角度来说，当一个线程从上到下执行一个需要加分布式锁的业务时，它首先需要进行加锁操作，当业务执行完毕后，再进行释放锁的操作。也就是先调用 tryLock() 函数再调用 releaseLock() 函数。
锁的可能误删操作会使得程序存在很严重的问题，所以需要实现加锁&解锁的归一化。
首先我们解释一下什么叫做加锁和解锁的归一化，简单来说，就是一个线程执行了加锁操作后，后续的解锁操作只能由该线程来执行，即加锁操作和解锁只能由同一线程来进行。
使用 ThreadLocal 和 UUID 来实现，代码如下：
public class RedisLockImpl implements RedisLock {
@Autowired
private StringRedisTemplate stringRedisTemplate;
private ThreadLocal<string> threadlocal = new ThreadLocal<>();
@Override
public boolean tryLock(String key, long timeout, TimeUnit unit) {
String uuid = UUID.randomUUID().toString();
threadlocal.set(uuid);
return stringRedisTemplate.opsForValue().setIfAbsent(key, uuid, timeout, unit);
}
@Override
public void releaseLock(String key) {
if (threadLocal.get().equals(stringRedisTemplate.opsForValue().get(key))) {
stringRedisTemplate.delete(key);
}
}
}
原理：ThreadLocal的get和set源码如下：
/**
* Returns the value in the current thread's copy of this
* thread-local variable. If the variable has no value for the
* current thread, it is first initialized to the value returned
* by an invocation of the {@link #initialValue} method.
*
* @return the current thread's value of this thread-local
*/
public T get() {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null) {
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
return result;
}
}
return setInitialValue();
}
/**
* Sets the current thread's copy of this thread-local variable
* to the specified value. Most subclasses will have no need to
* override this method, relying solely on the {@link #initialValue}
* method to set the values of thread-locals.
*
* @param value the value to be stored in the current thread's copy of
* this thread-local.
*/
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
}
/**
* Get the map associated with a ThreadLocal. Overridden in
* InheritableThreadLocal.
*
* @param t the current thread
* @return the map
*/
ThreadLocalMap getMap(Thread t) {
// 该值为Thread类中的成员变量，初始值为null：ThreadLocal.ThreadLocalMap threadLocals = null;
return t.threadLocals;
}
4）可重入发布式锁实现
上面的代码实现，可以保证当一个线程成功在 Redis 中设置了锁标志位后，其他线程再设置锁标志位时，返回 false。但是在一些场景下我们需要实现线程的重入，即相同的线程能够多次获取同一把锁，不需要等待锁释放后再去加锁。所以我们需要利用一些方式来实现分布式锁的可重入型，在 JDK 1.6 之后提供的内存级锁很多都支持可重入型，比如 synchronized 和 J.U.C 下的 Lock，其本质都是一样的，比对已经获得锁的线程是否与当前线程相同，是则重入，当释放锁时则需要根据重入的次数，来判断此时锁是否真正释放掉了。那么我们就按照这个思路来实现一个可重入的分布式锁：
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@Service
public class RedisLockImpl implements RedisLock {
@Autowired
private StringRedisTemplate stringRedisTemplate;
private static ThreadLocal<String> localString = new ThreadLocal<>();
private static ThreadLocal<Integer> localInteger = new ThreadLocal<>();
private static final Logger logger = LoggerFactory.getLogger(RedisLockImpl.class);
private static final long REENTRY_SPIN_SLEEP = 1000;
@Override
public boolean tryLock(String key, long timeout, TimeUnit unit) {
boolean isLock = false;
//
if (localString.get() == null) {
String uuid = UUID.randomUUID().toString();
localString.set(uuid);
while (true) {
isLock = Boolean.TRUE.equals(stringRedisTemplate.opsForValue().setIfAbsent(key, uuid, timeout, unit));
if (isLock) {
break;
}
try {
Thread.sleep(REENTRY_SPIN_SLEEP);
} catch (InterruptedException e) {
logger.error("线程中断异常");
throw new Exception("线程中断异常");
}
}
localInteger.set(0);
}
if(isLock) {
localInteger.set(localInteger.get() + 1);
}
return true;
}
@Override
public void releaseLock(String key) {
if (localString.get() != null
&& localString.get().equalsIgnoreCase(stringRedisTemplate.opsForValue().get(key))) {
if (localInteger.get() != null && localInteger.get() > 0) {
localInteger.set(localInteger.get() - 1);
}
} else {
stringRedisTemplate.delete(key);
localString.remove();
localInteger.remove();
}
}
}
5）分布式自旋锁实现
上面代码实现中，加入我们不能一次性获取到锁，那么就会直接返回失败，这对业务来说是十分不友好的，假设用户此时下单，刚好有另外一个用户也在下单，而且获取到了锁资源，那么该用户尝试获取锁之后失败，就只能直接返回“下单失败”的提示信息的。所以我们需要实现以自旋的形式来获取到锁，即不停的重试，基于这个想法，实现代码如下：
public class RedisLockImpl implements RedisLock {
@Autowired
private StringRedisTemplate stringRedisTemplate;
private ThreadLocal<String> threadLocal = new ThreadLocal<>();
private ThreadLocal<Integer> threadLocalInteger = new ThreadLocal<>();
@Override
public boolean tryLock(String key, long timeout, TimeUnit unit) {
Boolean isLocked = false;
if (threadLocal.get() == null) {
String uuid = UUID.randomUUID().toString();
threadLocal.set(uuid);
isLocked = stringRedisTemplate.opsForValue().setIfAbsent(key, uuid, timeout, unit);
// 尝试获取锁失败，则自旋获取锁直至成功
if (!isLocked) {
for (;;) {
isLocked = stringRedisTemplate.opsForValue().setIfAbsent(key, uuid, timeout, unit);
if (isLocked) {
break;
}
}
}
} else {
isLocked = true;
}
// 重入次数加1
if (isLocked) {
Integer count = threadLocalInteger.get() == null ? 0 : threadLocalInteger.get();
threadLocalInteger.set(count++);
}
return isLocked;
}
@Override
public void releaseLock(String key) {
// 判断当前线程所对应的uuid是否与Redis对应的uuid相同，再执行删除锁操作
if (threadLocal.get().equals(stringRedisTemplate.opsForValue().get(key))) {
Integer count = threadLocalInteger.get();
// 计数器减为0时才能释放锁
if (count == null || --count <= 0) {
stringRedisTemplate.delete(key);
}
}
}
}
6）超时问题
在高并发场景下，一把锁可能会被 N 多的进程竞争，获取锁后的业务代码也可能十分复杂，其运行时间可能偶尔会超过我们设置的过期时间，那么这个时候锁就会自动释放，而其他的进程就有可能来争抢这把锁，而此时原来获得锁的进程也在同时运行，这就有可能导致超卖现象或者其他并发安全问题。
那么如何解决这个问题呢？思路很简单，就是每隔一段时间去检查当前线程是否还在运行，如果还在运行，那么就继续更新锁的占有时长，而在释放锁的时候。具体的实现稍微复杂些，这里给出简易的代码实现： 类来执行更新锁超时的时间。
上述解决分布式锁失效的方案在分布式锁领域有一个专业的术语叫做 “异步续命” 。需要注意的是：当业务代码执行完毕后，我们需要停止更新锁超时时间的线程。
7）高并发
如果我们系统中利用 Redis 来实现分布式锁，而 Redis 的读写并发量约合 5 万左右。假设现在一个秒杀业务需要支持的并发量超过百万级别，那么如果这 100万的并发全部打入 Redis 中去请求锁资源，Redis 将会直接挂掉。所以我们现在应该来考虑如何解决这个问题，即如何在高并发的环境下保证 Redis 实现的分布式锁的可用性，接下来我们就来考虑一下这个问题。
在高并发的商城系统中，如果采用 Redis 缓存数据，则 Redis 缓存的并发能力是关键，因为很多的前缀操作都需要访问 Redis。而异步削峰只是基本操作，关键还是要保证 Redis 的并发处理能力。
解决这个问题的关键思想就是：分而治之，将商品库存分开放。
我们在 Redis 中存储商品的库存数量时，可以将商品的库存进行“分割”存储来提升 Redis 的读写并发量。
例如，原来的商品的 id 为 10001，库存为1000件，在Redis中的存储为(10001, 1000)，我们将原有的库存分割为5份，则每份的库存为200件，此时，我们在Redis 中存储的信息为(10001_0, 200)，(10001_1, 200)，(10001_2, 200)，(10001_3, 200)，(10001_4, 200)。
此时，我们将库存进行分割后，每个分割的库存使用商品 id 加上一个数字标识来存储，这样，在对存储商品库存的每个 key 进行 Hash 运算时，得出的 Hash 结果是不同的，这就说明，存储商品库存的 Key 有很大概率不在 Redis 的同一个槽位中，这就能够提升 Redis 处理请求的性能和并发量。
分割库存后，我们还需要在 Redis 中存储一份商品 ID 和 分割库存后的 Key 的映射关系，此时映射关系的 Key 为商品的 ID，也就是 10001，Value 为分割库存后存储库信息的 Key，也就是 10001_0，10001_1，10001_2，10001_3，10001_4。在 Redis 中我们可以使用 List 来存储这些值。
在真正处理库存信息时，我们可以先从 Redis 中查询出商品对应的分割库存后的所有 Key，同时使用 AtomicLong 来记录当前的请求数量，使用请求数量对从Redis 中查询出的商品对应的分割库存后的所有Key的长度进行求模运算，得出的结果为0，1，2，3，4。再在前面拼接上商品id就可以得出真正的库存缓存的Key。此时，就可以根据这个Key直接到Redis中获取相应的库存信息。
同时，我们可以将分隔的不同的库存数据分别存储到不同的 Redis 服务器中，进一步提升 Redis 的并发量。