redisson分布式锁学习

2023-07-31 | 阅读：次

什么是分布式锁?
传统redis工具类实现分布式锁 - {:.} 死锁问题 - {:.} 锁时间不合理问题 - {:.} 功能单一问题
redisson分布式锁 - {:.} 什么是redisson分布式锁？ - {:.} 如何解决传统redis工具类问题？ - {:.} 死锁问题 - {:.} 锁时间不合理问题 - {:.} 功能单一问题 - {:.} redisson加锁流程 - {:.} 看门狗机制 - {:.}
什么是看门狗机制？ - {:.} 什么时候会启动看门狗机制？ - {:.} 看门狗机制源码分析

什么是分布式锁?

当有多个线程并发访问同一共享数据时,如果多个线程同时都去修改这个共享数据,且修改操作不是原子操作,就很有可能出现线程安全问题，而产生线程安全问题的根本原因是缺乏对共享数据访问的同步和互斥。
为了解决这个问题，通常我们的做法是通过加锁来解决该问题,比如ReentrantLock or Synchronized ,但是在分布式系统中,存在多台服务器与客户端,这些节点之间都可能访问相同的共享数据。而Java中的内置锁机制如synchronized和ReentrantLock都是JVM内部的,无法对其他服务器产生效果。因此无法解决真正的分布式多线程访问安全问题。
为了实现分布式环境下的线程安全,需要引入外部的协调组件,实现一个分布式锁，常见的分布式锁组件有Redis、Zookeeper等，当然也可以基于数据库的悲观锁或CAS操作实现分布式锁。

传统redis工具类实现分布式锁

通过redis工具类基于实现分布式锁。

/**
* 加锁
*
* @param key               - key名称
* @param expireMillisecond - 锁成功后的有效期,毫秒
* @return return null or empty string is lock failed Otherwise return uuid value of lock-key
*/
public String lock(String key, long expireMillisecond) {
    Preconditions.checkArgument(StringUtils.isNotBlank(key));
    Preconditions.checkArgument(expireMillisecond > 0L);

    String lockKey = LOCK_KEY_PREFIX + key;
    String lockValue = UUID.randomUUID().toString();
    boolean keySet = redisTemplateWarpper.vSetIfAbsent(lockKey, lockValue, expireMillisecond);
    if (keySet) {   //锁成功
        return lockValue;
    }
    return null;
}

/**
* 解锁
*
* @param key
*/
public void unlock(String key, String value) {
    if (StringUtils.isBlank(value)) {
        return;
    }

    String lockKey = LOCK_KEY_PREFIX + key;
    String lockValueRedis = redisTemplateWarpper.vGet(lockKey);
    if (StringUtils.equals(lockValueRedis, value)) {
        redisTemplateWarpper.kDelete(lockKey);
    }
}

public Boolean vSetIfAbsent(String key, String value, long timeoutMillisecond) {
        RedisSerializer<String> stringSerializer = stringRedisTemplate.getStringSerializer();
        return stringRedisTemplate.execute(new RedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
                Object obj = connection.execute("set",
                        stringSerializer.serialize(checkKey(key)),
                        stringSerializer.serialize(value),
                        stringSerializer.serialize("NX"),
                        stringSerializer.serialize("PX"),
                        stringSerializer.serialize(String.valueOf(timeoutMillisecond)));
                return obj != null;
            }
        });
    }

传统的工具类实现redis分布式锁实现方式简单，虽然可以提供分布式锁的效果，但实际效果其实并不理想，因为在特殊情况下存在种种问题。

死锁问题

业务阻塞死锁： 某个客户端在执行一个长时间的阻塞操作，例如使用 BLPOP 或 BRPOP 命令来阻塞地等待列表中的元素。如果该操作长时间未完成或未释放连接，其他客户端可能无法获取连接，导致死锁。

客户端宕机死锁：客户端在解锁前崩溃下线,未设置过期时间，导致锁无法释放。

锁时间不合理问题

在锁资源的时候我们可以给锁设置过期时间，但锁的时间过长或过短都会出现问题。
例如：锁时间过长，其他线程一直无法获取锁资源，从而阻塞业务，不能够正常进行。
锁时间过短，锁资源中的任务还未执行完毕，其他线程已经可以获取到锁资源，从而操作共享数据，出现线程安全问题。

功能单一问题

仅能提供基础的加锁与解锁的功能，高级功能需要自己实现(例如读写锁、公平锁等等)

redisson分布式锁

什么是redisson分布式锁？

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) **

如何解决传统redis工具类问题？

死锁问题

解决业务阻塞死锁：通过tryLock(long waitTime, TimeUnit unit) 尝试获取锁，其他线程一定时间未获取不到锁就返回false，停止获取锁。
解决客户端宕机死锁：不传过期时间时，默认会设置30秒的过期时间，节点没宕机的情况下如果任务未执行完会持续进行锁续期，节点宕机后则不会再进行续期，到了过期时间后就会删掉key

锁时间不合理问题

通过看门狗机制自动进行锁续期，不进行人工干预。

功能单一问题

不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法

redisson加锁流程

看门狗机制

什么是看门狗机制？

Redisson提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期，也就是说，如果一个拿到锁的线程一直没有完成逻辑，那么看门狗会帮助线程不断的延长锁超时时间，锁不会因为超时而被释放。
默认情况下，看门狗的续期时间是30s，也可以通过修改Config.lockWatchdogTimeout来另行指定。
另外Redisson 还提供了可以指定leaseTime参数的加锁方法来指定加锁的时间。超过这个时间后锁便自动解开了，不会延长锁的有效期。

什么时候会启动看门狗机制？

方法	描述	Watch Dog 延期机制
lock()	拿锁失败时会不停的重试，直到成功获取锁	有，续锁时间默认为30秒，每隔30/3=10秒续锁
tryLock(10, TimeUnit.SECONDS)	尝试在10秒内获取锁，获取成功返回true，失败返回false	有，续锁时间默认为30秒
lock(10, TimeUnit.SECONDS) （void lock(long leaseTime, TimeUnit unit);）	拿锁失败时会不停的重试，10秒后自动释放锁	无，10秒后自动释放锁
tryLock(100, 10, TimeUnit.SECONDS) （boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) ）	尝试在100秒内获取锁，每次重试间隔为10秒，获取成功返回true，失败返回false	无，10秒后自动释放锁

如果你想让Redisson启动看门狗机制，你就不能自己在获取锁的时候，定义超时释放锁的时间
无论是通过lock() 还是通过tryLock获取锁，只要在参数中，不传入releastime，就会开启看门狗机制。
就是这两个方法不要用：
boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException
void lock(long leaseTime, TimeUnit unit);
因为它俩都传release，但是，你传的leaseTime是-1，也是会开启看门狗机制的

看门狗机制源码分析

 // 直接使用lock无参数方法
public void lock() {
    try {
        lock(-1, null, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

// 进入该方法 其中leaseTime = -1
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return;
    }

   //...
}

// 进入 tryAcquire(-1, leaseTime, unit, threadId)
private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}

// 进入 tryAcquireAsync
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        RFuture<Long> ttlRemainingFuture;
    	//  leaseTime = -1
        if (leaseTime > 0) {
            ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        } else {
            ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                    TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        }
        CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);
        ttlRemainingFuture = new CompletableFutureWrapper<>(s);

        CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
            // lock acquired leaseTime = -1
            if (ttlRemaining == null) {
                if (leaseTime > 0) {
                    internalLockLeaseTime = unit.toMillis(leaseTime);
                } else {
            	// 看门狗续期
                    scheduleExpirationRenewal(threadId);
                }
            }
            return ttlRemaining;
        });
        return new CompletableFutureWrapper<>(f);
    }

 protected void scheduleExpirationRenewal(long threadId) {
        ExpirationEntry entry = new ExpirationEntry();
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        if (oldEntry != null) {
            oldEntry.addThreadId(threadId);
        } else {
            entry.addThreadId(threadId);
            try {
                renewExpiration();
            } finally {
                if (Thread.currentThread().isInterrupted()) {
                    cancelExpirationRenewal(threadId);
                }
            }
        }
    }


   private void renewExpiration() {
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ee == null) {
            return;
        }
        
        Timeout task = commandExecutor.getServiceManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
                if (ent == null) {
                    return;
                }
                Long threadId = ent.getFirstThreadId();
                if (threadId == null) {
                    return;
                }
                
                CompletionStage<Boolean> future = renewExpirationAsync(threadId);
                future.whenComplete((res, e) -> {
                    if (e != null) {
                        log.error("Can't update lock {} expiration", getRawName(), e);
                        EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                        return;
                    }
                    
                    if (res) {
                        // reschedule itself
                        renewExpiration();
                    } else {
                        cancelExpirationRenewal(null);
                    }
                });
            }
        // 默认锁租期internalLockLeaseTime = 30s  默认续期时间为锁租期/3 = 10s
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
        
        ee.setTimeout(task);
    }