Redis分布式代理锁的两种实现

Redis分布式代理锁的两种实现

一、引言

        分布式环境中的业务开发，Synchronized、ReentrantLock等本地锁已经不能再防止并发冲突，分布式锁应运而生。Redis分布式锁是目前最火热的锁工具之一，但是项目中对于并发的控制加锁解锁非常频繁，冗余代码较多，锁管理分散。

        本文基于Redisson通过两种方式实现代理分布式锁：

                1、ThreadLocal线程缓存 + AOP切面

                2、AOP切面 + 入参固定

二、未使用代理组件

未使用代理组建的情况下，一旦需要加锁都会进行以下编码，除了业务处理其他代码在业务功能中随处可见，冗余了许多非必要代码。

String lockKey = RedisConsts.ORDER_ID_LOCK + orderId;
RLock lock = Lock(lockKey);
try {if (Lock(RedisConsts.WAIT, RedisConsts.EXECUTE, RedisConsts.TIME_UNIT)) {//订单处理} else {("加锁失败:{}",lockKey);}
} catch (InterruptedException e) {("获取锁异常:{}", e);throw new BusinessException(ErrorCodeEnum.FAILED_TO_ACQUIRE_LOCK);
} finally {if (lock.isHeldByCurrentThread()) {lock.unlock();}
}

三、使用代理组件

1、注解

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RedisLock {String key();long waitOut() default 60;long executeOut() default 60;TimeUnit timeUnit() default TimeUnit.SECONDS;//是否自动清除LockUtilboolean atuoRemove() default false;String suffixKeyTypeEnum() default RedisLockCommonUtil.THREAD_LOCAL;String objectName() default "";String[] paramName() default {};
}

2、代理锁工具类

public class LockUtil {static ThreadLocal<String> LOCK_KEY = new ThreadLocal<String>();public static void set(String key) {LOCK_KEY.set(key);}public static String get() {return ();}public static void remove() {ve();}}

3、切面

个人更喜欢使用ThreadLocal。

使用固定入参需要解析参数，会增加一些性能消耗，虽然对于越来越快的计算性能，这点消耗可以忽略。

① 使用ThreadLocal（适用于责任链、工厂模式等长链路模式）

@Slf4j
@Aspect
@Component
public class RedisLockAspect {@Resourceprivate RedissonClient redissonClient;@Pointcut("@annotation(***.dislock.annotation.RedisLock)")public void lockPointCut() {}@Around("lockPointCut() && @annotation(redisLock)")public Object around(ProceedingJoinPoint joinPoint, RedisLock redisLock) {String lockKey = redisLock.key();if (())) {//获取线程缓存中的锁后缀lockKey += ();}log.info("开始代理加锁:{}",lockKey);RLock lock = Lock(lockKey);try {if (Lock(redisLock.waitOut(), uteOut(), redisLock.timeUnit())) {log.info("代理加锁成功:{}",lockKey);return joinPoint.proceed();} else {log.warn("代理加锁失败:{}",lockKey);}} catch (InterruptedException e) {("获取代理锁异常:{}", e);throw new BusinessException(ErrorCodeEnum.FAILED_TO_ACQUIRE_LOCK);} finally {if (lock.isHeldByCurrentThread()) {lock.unlock();log.info("代理解锁:{}",lockKey);}//如果方法注解中开启自动清除，就去除if (redisLock.atuoRemove()) {ve();log.info("自动清除LockUtil:{}",lockKey);}}return null;}}

② 使用固定入参（适用于分散型方法）

@Slf4j
@Aspect
@Component
public class RedisLockAspect {@Resourceprivate RedissonClient redissonClient;@Pointcut("@annotation(***.dislock.annotation.RedisLock)")public void lockPointCut() {}//定义固定入参名，业务使用时需要定义相同的参数名并传入下游方法private final static String REDIS_KEY = "redisKey";@Around("lockPointCut() && @annotation(redisLock)")public void around(ProceedingJoinPoint joinPoint, RedisLock redisLock) {String objectName = redisLock.objectName();if (StringUtil.isBlank(objectName)) {throw new BusinessException("objectName为空");}String[] paramName = redisLock.paramName();Object[] args = Args();String[] objectNames = ((CodeSignature) Signature()).getParameterNames();Map<String, Object> objectHashMap = new HashMap<>();for (int i = 0; i < objectNames.length; i++) {objectHashMap.put(objectNames[i], args[i]);}if (!ainsKey(objectName)) {throw new BusinessException("入参不包含该对象" + objectName);}Object o = (objectName);if (paramName == null || paramName.length == 0) {return redisLock.key() + o;}String lockKey = redisLock.key();for (int i = 0; i < paramName.length; i++) {lockKey += FieldValueByName(paramName[i], o);}log.info("开始代理加锁:{}",lockKey);RLock lock = Lock(lockKey);try {if (Lock(redisLock.waitOut(), uteOut(), redisLock.timeUnit())) {log.info("代理加锁成功:{}",lockKey);joinPoint.proceed();}} catch (InterruptedException e) {("获取代理锁异常:{}", e);throw new BusinessException(ErrorCodeEnum.FAILED_TO_ACQUIRE_LOCK);} finally {if (lock.isHeldByCurrentThread()) {lock.unlock();log.info("代理解锁:{}",lockKey);}}}
}

三、DEMO测试

1、模拟并发，两个可以同时进行，第三个线程有锁争抢

Thread lock1 = new Thread(new Runnable() {@Overridepublic void run() {snChainDTO1.setSn("666sn");LockUtil.Sn());lockProcess.process(snChainDTO1);}}
);
lock1.start();
log.info("lock1 start");
MacSnChainDTO snChainDTO2 = new MacSnChainDTO();
Thread lock2 = new Thread(new Runnable() {@Overridepublic void run() {snChainDTO2.setSn("888sn");LockUtil.Sn());lockProcess.process(snChainDTO2);}}
);
lock2.start();
log.info("lock2 start");
MacSnChainDTO snChainDTO3 = new MacSnChainDTO();
Thread lock3 = new Thread(new Runnable() {@Overridepublic void run() {snChainDTO3.setSn("666sn");LockUtil.Sn());lockProcess.process(snChainDTO3);}}
);
lock3.start();
log.info("lock3 start");

 2、测试业务方法

@RedisLock(key = RedisPrefixConstant.DAMAGE_SN_LOCK_KEY, atuoRemove = true)
public void process(MacSnChainDTO macSnChainDTO) {for (int i = 0; i < 6; i++) {log.info("测试加锁:{}", () + i);}
}

3、测试结果

2022-04-11 16:02:58.027 bit [http-nio-8189-exec-1] INFO  c.stlock.LockEnterImpl.lockTogether:58 - lock1 start
2022-04-11 16:02:58.028 bit [http-nio-8189-exec-1] INFO  c.stlock.LockEnterImpl.lockTogether:74 - lock2 start
2022-04-11 16:02:58.029 bit [Thread-98] INFO  c.dislock.aspect.RedisLockAspect.around:36 - 开始代理加锁:BIT:LOCK:DAMAGE:SN:888sn
2022-04-11 16:02:58.029 bit [Thread-97] INFO  c.dislock.aspect.RedisLockAspect.around:36 - 开始代理加锁:BIT:LOCK:DAMAGE:SN:666sn
2022-04-11 16:02:58.030 bit [http-nio-8189-exec-1] INFO  c.stlock.LockEnterImpl.lockTogether:90 - lock3 start
2022-04-11 16:02:58.031 bit [Thread-99] INFO  c.dislock.aspect.RedisLockAspect.around:36 - 开始代理加锁:BIT:LOCK:DAMAGE:SN:666sn
2022-04-11 16:02:58.033 bit [http-nio-8189-exec-1] INFO  c.stlock.LockEnterImpl.logAround:52 - method: lockTogether, result: {"success":true}, span: 23
2022-04-11 16:02:58.063 bit [Thread-97] INFO  c.dislock.aspect.RedisLockAspect.around:40 - 代理加锁成功:BIT:LOCK:DAMAGE:SN:666sn
2022-04-11 16:02:58.066 bit [Thread-98] INFO  c.dislock.aspect.RedisLockAspect.around:40 - 代理加锁成功:BIT:LOCK:DAMAGE:SN:888sn
2022-04-11 16:02:58.067 bit [Thread-98] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:888sn0
2022-04-11 16:02:58.067 bit [Thread-97] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn0
2022-04-11 16:02:58.068 bit [Thread-98] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:888sn1
2022-04-11 16:02:58.068 bit [Thread-97] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn1
2022-04-11 16:02:58.068 bit [Thread-98] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:888sn2
2022-04-11 16:02:58.068 bit [Thread-97] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn2
2022-04-11 16:02:58.068 bit [Thread-98] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:888sn3
2022-04-11 16:02:58.068 bit [Thread-97] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn3
2022-04-11 16:02:58.068 bit [Thread-97] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn4
2022-04-11 16:02:58.068 bit [Thread-98] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:888sn4
2022-04-11 16:02:58.068 bit [Thread-97] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn5
2022-04-11 16:02:58.068 bit [Thread-98] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:888sn5
2022-04-11 16:02:58.084 bit [Thread-97] INFO  c.dislock.aspect.RedisLockAspect.around:48 - 代理解锁:BIT:LOCK:DAMAGE:SN:666sn
2022-04-11 16:02:58.084 bit [Thread-97] INFO  c.stlock.LockEnterImpl.run:51 - lock1-LockUtil:666sn
2022-04-11 16:02:58.084 bit [Thread-97] INFO  c.stlock.LockEnterImpl.run:53 - 自动清除LockUti:BIT:LOCK:DAMAGE:SN:666sn
2022-04-11 16:02:58.090 bit [Thread-98] INFO  c.dislock.aspect.RedisLockAspect.around:48 - 代理解锁:BIT:LOCK:DAMAGE:SN:888sn
2022-04-11 16:02:58.091 bit [Thread-98] INFO  c.stlock.LockEnterImpl.run:67 - lock2-LockUtil:888sn
2022-04-11 16:02:58.091 bit [Thread-98] INFO  c.stlock.LockEnterImpl.run:69 - l自动清除LockUti:BIT:LOCK:DAMAGE:SN:888sn
2022-04-11 16:02:58.121 bit [Thread-99] INFO  c.dislock.aspect.RedisLockAspect.around:40 - 代理加锁成功:BIT:LOCK:DAMAGE:SN:666sn
2022-04-11 16:02:58.122 bit [Thread-99] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn0
2022-04-11 16:02:58.122 bit [Thread-99] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn1
2022-04-11 16:02:58.122 bit [Thread-99] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn2
2022-04-11 16:02:58.122 bit [Thread-99] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn3
2022-04-11 16:02:58.122 bit [Thread-99] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn4
2022-04-11 16:02:58.122 bit [Thread-99] INFO  c.stlock.LockProcessImpl.process:25 - 测试加锁:666sn5
2022-04-11 16:02:58.142 bit [Thread-99] INFO  c.dislock.aspect.RedisLockAspect.around:48 - 代理解锁:BIT:LOCK:DAMAGE:SN:666sn
2022-04-11 16:02:58.142 bit [Thread-99] INFO  c.stlock.LockEnterImpl.run:83 - lock3-LockUtil:666sn
2022-04-11 16:02:58.142 bit [Thread-99] INFO  c.stlock.LockEnterImpl.run:85 - 自动清除LockUti:BIT:LOCK:DAMAGE:SN:666sn

四、原理分析

1、ThreadLocal

        线程缓存是java提供的绑定当前线程的存储空间，因此在任意方法进行ThreadLocal 的设置，后续只要属于当前线程的方法都可以取到这个值。

        首先通过它的set方法分析

    public void set(T value) {Thread t = Thread.currentThread();//获取当前线程为键对应的ThreadLocalMapThreadLocalMap map = getMap(t);//将当前ThreadLocal对象作为基础，业务值设置到ThreadLocalMap维护的数组中if (map != null)map.set(this, value);elsecreateMap(t, value);}void createMap(Thread t, T firstValue) {//创建当前线程的ThreadLocalMapt.threadLocals = new ThreadLocalMap(this, firstValue);}ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {//ThreadLocalMap维护了一个数组，这是因为线程的缓存值可能有多个，许多第三方框架都在使用table = new Entry[INITIAL_CAPACITY];int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);table[i] = new Entry(firstKey, firstValue);size = 1;setThreshold(INITIAL_CAPACITY);}private void set(ThreadLocal<?> key, Object value) {Entry[] tab = table;int len = tab.length;//通过hasn值与数组长度取&获取下标int i = key.threadLocalHashCode & (len-1);//将业务值设置到数组中for (Entry e = tab[i];e != null;e = tab[i = nextIndex(i, len)]) {ThreadLocal<?> k = e.get();if (k == key) {e.value = value;return;}if (k == null) {replaceStaleEntry(key, value, i);return;}}tab[i] = new Entry(key, value);int sz = ++size;if (!cleanSomeSlots(i, sz) && sz >= threshold)rehash();}

        接着通过get方法分析

    public T get() {//获取当前线程Thread t = Thread.currentThread();//获取当前线程为键对应的ThreadLocalMap ThreadLocalMap map = getMap(t);if (map != null) {//将当前ThreaadLocal对象传入getEntry方法，获取线程缓存的值ThreadLocalMap.Entry e = Entry(this);if (e != null) {@SuppressWarnings("unchecked")T result = (T)e.value;return result;}}return setInitialValue();}private Entry getEntry(ThreadLocal<?> key) {//根据对象的hash值与存储数组进行&，获取存储值的下标int i = key.threadLocalHashCode & (table.length - 1);Entry e = table[i];if (e != null && e.get() == key)return e;elsereturn getEntryAfterMiss(key, i, e);}

        用完记得remove！由于ThreadLocalMap的Entry是虚引用，线程如果不销毁不会被回收，用完就进行remove会避免内存泄漏风险。

    public void remove() {ThreadLocalMap m = getMap(Thread.currentThread());if (m != ve(this);}private void remove(ThreadLocal<?> key) {Entry[] tab = table;int len = tab.length;int i = key.threadLocalHashCode & (len-1);for (Entry e = tab[i];e != null;e = tab[i = nextIndex(i, len)]) {if (e.get() == key) {//将虚引用置为nulle.clear();//清理数组expungeStaleEntry(i);return;}}}private int expungeStaleEntry(int staleSlot) {Entry[] tab = table;int len = tab.length;//将该下标对象置为null，便于gc回收tab[staleSlot].value = null;tab[staleSlot] = null;size--;Entry e;int i;//遍历该下标之后所有不为空的ThreadLocal对象for (i = nextIndex(staleSlot, len);(e = tab[i]) != null;i = nextIndex(i, len)) {ThreadLocal<?> k = e.get();if (k == null) {//如果该ThreadLocal实例的虚引用已经被销毁，将该位置的ThreadLocal置为nulle.value = null;tab[i] = null;size--;} else {int h = k.threadLocalHashCode & (len - 1);if (h != i) {tab[i] = null;while (tab[h] != null)h = nextIndex(h, len);tab[h] = e;}}}return i;}

2、AOP

        对于动态代理可以说是老生常谈了，这里就不进行描述了，做个字节码反编译给大家看看代理类的实际模样吧。

        可以看到代理类继承自Proxy，并且将LockProcessImpl 的方法编织成了接口方法，再由代理类实现接口方法。最终调用invoke方法，invoke方法在会通过反射获取到原始方法要执行的代码，并且进行方法增强。

        总结起来就是AOP将注解下的类方法进行包装，切面的代码将业务方法包在中间进行增强执行。

stlock.LockProcessImpl;
sponsibility.MacSnChainDTO;
import flect.InvocationHandler;
import flect.Method;
import flect.Proxy;
import flect.UndeclaredThrowableException;public final class LockProcessImpl extends Proxy implements LockProcessImpl {private static Method m1;private static Method m8;private static Method m3;private static Method m2;private static Method m6;private static Method m5;private static Method m7;private static Method m9;private static Method m0;private static Method m4;public LockProcessImpl(InvocationHandler paramInvocationHandler) {super(paramInvocationHandler);}public final boolean equals(Object paramObject) {try {return ((Boolean)this.h.invoke(this, m1, new Object[] { paramObject })).booleanValue();} catch (Error|RuntimeException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final void notify() {try {this.h.invoke(this, m8, null);return;} catch (Error|RuntimeException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final void process(MacSnChainDTO paramMacSnChainDTO) {try {this.h.invoke(this, m3, new Object[] { paramMacSnChainDTO });return;} catch (Error|RuntimeException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final String toString() {try {return (String)this.h.invoke(this, m2, null);} catch (Error|RuntimeException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final void wait(long paramLong) throws InterruptedException {try {this.h.invoke(this, m6, new Object[] { Long.valueOf(paramLong) });return;} catch (Error|RuntimeException|InterruptedException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final void wait(long paramLong, int paramInt) throws InterruptedException {try {this.h.invoke(this, m5, new Object[] { Long.valueOf(paramLong), Integer.valueOf(paramInt) });return;} catch (Error|RuntimeException|InterruptedException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final Class getClass() {try {return (Class)this.h.invoke(this, m7, null);} catch (Error|RuntimeException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final void notifyAll() {try {this.h.invoke(this, m9, null);return;} catch (Error|RuntimeException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final int hashCode() {try {return ((Integer)this.h.invoke(this, m0, null)).intValue();} catch (Error|RuntimeException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }public final void wait() throws InterruptedException {try {this.h.invoke(this, m4, null);return;} catch (Error|RuntimeException|InterruptedException error) {throw null;} catch (Throwable throwable) {throw new UndeclaredThrowableException(throwable);} }static {try {m1 = Class.forName("java.lang.Object").getMethod("equals", new Class[] { Class.forName("java.lang.Object") });m8 = Class.forName(&#stlock.LockProcessImpl").getMethod("notify", new Class[0]);m3 = Class.forName(&#stlock.LockProcessImpl").getMethod("process", new Class[] { Class.forName(&#sponsibility.MacSnChainDTO") });m2 = Class.forName("java.lang.Object").getMethod("toString", new Class[0]);m6 = Class.forName(&#stlock.LockProcessImpl").getMethod("wait", new Class[] { long.class });m5 = Class.forName(&#stlock.LockProcessImpl").getMethod("wait", new Class[] { long.class, int.class });m7 = Class.forName(&#stlock.LockProcessImpl").getMethod("getClass", new Class[0]);m9 = Class.forName(&#stlock.LockProcessImpl").getMethod("notifyAll", new Class[0]);m0 = Class.forName("java.lang.Object").getMethod("hashCode", new Class[0]);m4 = Class.forName(&#stlock.LockProcessImpl").getMethod("wait", new Class[0]);return;} catch (NoSuchMethodException noSuchMethodException) {throw new Message());} catch (ClassNotFoundException classNotFoundException) {throw new Message());} }
}