Redis分布式锁的正确实现集群版

为了确保分布式锁可用&＃xff0c;我们至少要确保锁的实现同时满足以下四个条件&＃xff1a;
1、互斥性。在任意时刻&＃xff0c;只有一个客户端能持有锁。
2、不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁&＃xff0c;也能保证后续其他客户端能加锁。
3、具有容错性。只要大部分的Redis节点正常运行&＃xff0c;客户端就可以加锁和解锁。
4、解铃还须系铃人。加锁和解锁必须是同一个客户端&＃xff0c;客户端自己不能把别人加的锁给解了。

package com.hz.tgb.data.redis.lock;import cn.hutool.core.util.IdUtil;
import com.hz.tgb.entity.Book;
import com.hz.tgb.spring.SpringUtils;
import io.lettuce.core.ScriptOutputType;
import io.lettuce.core.SetArgs;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.TimeoutUtils;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Supplier;/*** Redis分布式锁 - 集群版** &＃64;author hezhao on 2019.11.13*/
&＃64;Component
public class RedisClusterLockUtil {/*为了确保分布式锁可用&＃xff0c;我们至少要确保锁的实现同时满足以下四个条件&＃xff1a;1、互斥性。在任意时刻&＃xff0c;只有一个客户端能持有锁。2、不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁&＃xff0c;也能保证后续其他客户端能加锁。3、具有容错性。只要大部分的Redis节点正常运行&＃xff0c;客户端就可以加锁和解锁。4、解铃还须系铃人。加锁和解锁必须是同一个客户端&＃xff0c;客户端自己不能把别人加的锁给解了。*/private static final Logger logger &＃61; LoggerFactory.getLogger(RedisLockUtil.class);private static RedisTemplate cacheTemplate;/** OK: Redis操作是否成功 */private static final String REDIS_OK &＃61; "OK";/** CONN_NOT_FOUND: Redis链接类型不匹配 */private static final String REDIS_CONN_NOT_FOUND &＃61; "CONN_NOT_FOUND";/** 解锁是否成功 */private static final Long RELEASE_SUCCESS &＃61; 1L;/** 解锁Lua脚本 */private static final String UNLOCK_LUA_SCRIPT &＃61; "if redis.call(&＃39;get&＃39;, KEYS[1]) &＃61;&＃61; ARGV[1] then return redis.call(&＃39;del&＃39;, KEYS[1]) else return 0 end";/*** The number of nanoseconds for which it is faster to spin* rather than to use timed park. A rough estimate suffices* to improve responsiveness with very short timeouts.*/private static final long spinForTimeoutThreshold &＃61; 1000000L;/*** 加锁* &＃64;param lockKey 锁键* &＃64;param requestId 请求唯一标识* &＃64;param expireTime 缓存过期时间* &＃64;param unit 时间单位* &＃64;return true: 加锁成功, false: 加锁失败*/&＃64;SuppressWarnings("all")public static boolean lock(String lockKey, String requestId, long expireTime, TimeUnit unit) {// 加锁和设置过期时间必须是原子操作&＃xff0c;否则在高并发情况下或者Redis突然崩溃会导致数据错误。try {// 以毫秒作为过期时间long millisecond &＃61; TimeoutUtils.toMillis(expireTime, unit);String result &＃61; execute(connection -> {Object nativeConnection &＃61; connection.getNativeConnection();RedisSerializer keySerializer &＃61; (RedisSerializer) getRedisTemplate().getKeySerializer();RedisSerializer valueSerializer &＃61; (RedisSerializer) getRedisTemplate().getValueSerializer();// springboot 2.0以上的spring-data-redis 包默认使用 lettuce连接包// lettuce连接包下序列化键值&＃xff0c;否知无法用默认的ByteArrayCodec解析byte[] keyByte &＃61; keySerializer.serialize(lockKey);byte[] valueByte &＃61; valueSerializer.serialize(requestId);//lettuce连接包&＃xff0c;单机模式&＃xff0c;ex为秒&＃xff0c;px为毫秒if (nativeConnection instanceof RedisAsyncCommands) {RedisAsyncCommands commands &＃61; (RedisAsyncCommands)nativeConnection;// 同步方法执行、setnx禁止异步return commands.getStatefulConnection().sync().set(keyByte, valueByte, SetArgs.Builder.nx().px(millisecond));} else if (nativeConnection instanceof RedisAdvancedClusterAsyncCommands) {// lettuce连接包&＃xff0c;集群模式&＃xff0c;ex为秒&＃xff0c;px为毫秒RedisAdvancedClusterAsyncCommands clusterAsyncCommands &＃61; (RedisAdvancedClusterAsyncCommands) nativeConnection;return clusterAsyncCommands.getStatefulConnection().sync().set(keyByte, valueByte, SetArgs.Builder.nx().px(millisecond));}return REDIS_CONN_NOT_FOUND;});// 如果链接类型匹配不上&＃xff0c;使用默认加锁方法if (Objects.equals(result, REDIS_CONN_NOT_FOUND)) {return getRedisTemplate().opsForValue().setIfAbsent(lockKey, requestId)&& getRedisTemplate().expire(lockKey, expireTime, unit);}return REDIS_OK.equals(result);} catch (Exception e) {logger.error("RedisLockUtil lock 加锁失败", e);}return false;}/*** 解锁* &＃64;param lockKey 锁键* &＃64;param requestId 请求唯一标识* &＃64;return true: 解锁成功, false: 解锁失败*/&＃64;SuppressWarnings("all")public static boolean unLock(String lockKey, String requestId) {try {// 使用Lua脚本实现解锁的原子性&＃xff0c;如果requestId相等则解锁Object result &＃61; execute(connection -> {Object nativeConnection &＃61; connection.getNativeConnection();RedisSerializer keySerializer &＃61; (RedisSerializer) getRedisTemplate().getKeySerializer();RedisSerializer valueSerializer &＃61; (RedisSerializer) getRedisTemplate().getValueSerializer();// springboot 2.0以上的spring-data-redis 包默认使用 lettuce连接包// lettuce连接包下序列化键值&＃xff0c;否知无法用默认的ByteArrayCodec解析byte[] keyByte &＃61; keySerializer.serialize(lockKey);byte[] valueByte &＃61; valueSerializer.serialize(requestId);//lettuce连接包&＃xff0c;单机模式if (nativeConnection instanceof RedisAsyncCommands) {RedisAsyncCommands commands &＃61; (RedisAsyncCommands)nativeConnection;// 同步方法执行、setnx禁止异步byte[][] keys &＃61; {keyByte};byte[][] values &＃61; {valueByte};return commands.getStatefulConnection().sync().eval(UNLOCK_LUA_SCRIPT, ScriptOutputType.INTEGER, keys , values);} else if (nativeConnection instanceof RedisAdvancedClusterAsyncCommands) {// lettuce连接包&＃xff0c;集群模式RedisAdvancedClusterAsyncCommands clusterAsyncCommands &＃61; (RedisAdvancedClusterAsyncCommands) nativeConnection;byte[][] keys &＃61; {keyByte};byte[][] values &＃61; {valueByte};return clusterAsyncCommands.getStatefulConnection().sync().eval(UNLOCK_LUA_SCRIPT, ScriptOutputType.INTEGER, keys , values);}return REDIS_CONN_NOT_FOUND;});// 如果链接类型匹配不上&＃xff0c;使用默认解锁方法if (Objects.equals(result, REDIS_CONN_NOT_FOUND)) {return getRedisTemplate().delete(lockKey);}return Objects.equals(RELEASE_SUCCESS, result);} catch (Exception e) {logger.error("RedisLockUtil unLock 解锁失败", e);}return false;}/*** 阻塞锁&＃xff0c;拿到锁后执行业务逻辑。注意&＃xff1a;超时返回null&＃xff0c;程序会继续往下执行* &＃64;param callback 业务处理逻辑&＃xff0c;入参默认为NULL* &＃64;param lockKey 锁键* &＃64;param timeout 超时时长, 缓存过期时间默认等于超时时长* &＃64;param unit 时间单位* &＃64;return R*/public static R tryLock(Supplier callback, String lockKey, long timeout, TimeUnit unit) {return tryLock(callback, lockKey, IdUtil.fastSimpleUUID(), timeout, timeout, unit, TimeOutProcess.DEFAULT);}/*** 阻塞锁&＃xff0c;拿到锁后执行业务逻辑。注意&＃xff1a;超时会抛出异常* &＃64;param callback 业务处理逻辑&＃xff0c;入参默认为NULL* &＃64;param lockKey 锁键* &＃64;param timeout 超时时长, 缓存过期时间默认等于超时时长* &＃64;param unit 时间单位* &＃64;return R*/public static R tryLockTimeout(Supplier callback, String lockKey, long timeout, TimeUnit unit) {return tryLock(callback, lockKey, IdUtil.fastSimpleUUID(), timeout, timeout, unit, TimeOutProcess.THROW_EXCEPTION);}/*** 阻塞锁&＃xff0c;拿到锁后执行业务逻辑。注意&＃xff1a;超时会给予补偿&＃xff0c;即处理正常逻辑* &＃64;param callback 业务处理逻辑&＃xff0c;入参默认为NULL* &＃64;param lockKey 锁键* &＃64;param timeout 超时时长, 缓存过期时间默认等于超时时长* &＃64;param unit 时间单位* &＃64;return R*/public static R tryLockCompensate(Supplier callback, String lockKey, long timeout, TimeUnit unit) {return tryLock(callback, lockKey, IdUtil.fastSimpleUUID(), timeout, timeout, unit, TimeOutProcess.CARRY_ON);}/*** 阻塞锁&＃xff0c;拿到锁后执行业务逻辑* &＃64;param callback 业务处理逻辑* &＃64;param lockKey 锁键* &＃64;param requestId 请求唯一标识* &＃64;param timeout 超时时长* &＃64;param expireTime 缓存过期时间* &＃64;param unit 时间单位* &＃64;param timeoutProceed 超时处理逻辑* &＃64;return R*/public static R tryLock(Supplier callback, String lockKey, String requestId,long timeout, long expireTime, TimeUnit unit, TimeOutProcess timeoutProceed) {boolean lockFlag &＃61; false;try {lockFlag &＃61; tryLock(lockKey, requestId, timeout, expireTime, unit);if(lockFlag){return callback.get();}} finally {if (lockFlag){unLock(lockKey, requestId);}}if (timeoutProceed &＃61;&＃61; null) {return null;}if (timeoutProceed &＃61;&＃61; TimeOutProcess.THROW_EXCEPTION) {throw new RedisLockTimeOutException();}if (timeoutProceed &＃61;&＃61; TimeOutProcess.CARRY_ON) {return callback.get();}return null;}/*** 阻塞锁* &＃64;param lockKey 锁键* &＃64;param requestId 请求唯一标识* &＃64;param timeout 超时时长, 缓存过期时间默认等于超时时长* &＃64;param unit 时间单位* &＃64;return true: 加锁成功, false: 加锁失败*/public static boolean tryLock(String lockKey, String requestId, long timeout, TimeUnit unit) {return tryLock(lockKey, requestId, timeout, timeout, unit);}/*** 阻塞锁* &＃64;param lockKey 锁键* &＃64;param requestId 请求唯一标识* &＃64;param timeout 超时时长* &＃64;param expireTime 缓存过期时间* &＃64;param unit 时间单位* &＃64;return true: 加锁成功, false: 加锁失败*/public static boolean tryLock(String lockKey, String requestId, long timeout, long expireTime, TimeUnit unit) {long nanosTimeout &＃61; unit.toNanos(timeout);if (nanosTimeout <&＃61; 0L) {return false;}final long deadline &＃61; System.nanoTime() &＃43; nanosTimeout;for (;;) {// 获取到锁if (lock(lockKey, requestId, expireTime, unit)) {return true;}// 判断是否需要继续阻塞&＃xff0c; 如果已超时则返回falsenanosTimeout &＃61; deadline - System.nanoTime();if (nanosTimeout <&＃61; 0L) {return false;}// 休眠1毫秒if (nanosTimeout > spinForTimeoutThreshold) {LockSupport.parkNanos(spinForTimeoutThreshold);}}}public static T execute(RedisCallback action) {return getRedisTemplate().execute(action);}public static RedisTemplate getRedisTemplate() {if (cacheTemplate &＃61;&＃61; null) {cacheTemplate &＃61; SpringUtils.getBean("redisTemplate", RedisTemplate.class);}return cacheTemplate;}public static void main(String[] args) {Book param &＃61; new Book();param.setBookId(1234);param.setName("西游记");Boolean flag &＃61; tryLock(() -> {int bookId &＃61; param.getBookId();System.out.println(bookId);// TODO ...return true;}, "BOOK-" &＃43; param.getBookId(), 3, TimeUnit.SECONDS);System.out.println(flag);}/*** 超时处理逻辑*/public enum TimeOutProcess {/** 默认&＃xff0c;超时返回null&＃xff0c;程序会继续往下执行 */DEFAULT,/** 超时会抛出异常 */THROW_EXCEPTION,/** 超时会给予补偿&＃xff0c;即处理正常逻辑 */CARRY_ON,}}