Java 分布式锁实现的一些实践

近期换工作，闲下来有点时间写点东西，在这里分享一些心得体会

背景：我们在做后端开发时，无法避免的会遇到一些一致性问题，有时候我们前端的小伙伴或者rpc接口的调用方，在很短的时间间隔内给我们相同的请求，由此可能会导致一些无法预见的问题，因此需要我们在接口层面处理，下面给大家分享一下我解决此类问题的一些实践

思路：1、加锁解锁的逻辑与正常的业务逻辑需要分开，不能耦合，否则会增加后期接口的维护成本，考虑使用自定义注解+aop；

　　　2、锁的实现方式有很多，根据我们不同的场景和条件有不同的选择，因此加锁和解锁需要进行抽象，aop中只依赖锁的接口；

　　 3、加锁时，我们需要生成key，用来区分是不是同一个请求，这里key的生成比较关键，不同的业务场景key的生成差别很大，我的想法是将这部分key生成逻辑抽象化，定义成接口，后面根据接口入参情况自己实现,当然也可以自己约定规则，通过反射去拼接；　

实现步骤：

1 自定义注解：

/**

 * 锁注解 标记需要加锁的接口

 * @author XuZhangxing

 */

@Target(ElementType.METHOD)

@Retention(RetentionPolicy.RUNTIME)

public @interface LockMask {

    // 锁超时释放时间

    long expireSecond() default 10;

    // 锁key的前缀

    LockKeyPrefix lockKeyPre() default LockKeyPrefix.WMD;

    // 方法参数位置默认是0 第一个

    int index() default 0;

    // el表达式 可以用来解析key

    String el() default "";

}

2 锁对象dto

@Data

public final class LockParam {

    // 锁的key

    private String key;

    // 针对一些特殊场景用

    private String val;

    // 超时时间 单位为秒

    private long expireSecond;

    // 加锁是否成功，用来在finally里面释放锁用

    private Boolean success = Boolean.FALSE;

    public LockParam(String key, long expireSecond) {

        this.key = key;

        this.expireSecond = expireSecond;

    }

    public LockParam() {

    }

    @Override

    public String toString() {

        return JsonUtil.obj2String(this);

    }

}

/**
 * 锁key前缀
 *  标记哪个业务系统
 *
 * @author XuZhangxing
 */
public enum LockKeyPrefix {
    WMD
}

3 key生成抽象接口

/**

 * 

 * key生成 为了简化操作，直接由调用方去实现key的生成

 *

 * @author XuZhangxing

 */

@FunctionalInterface

public interface KeyGenerate {

    // 给key使用的 list 支持粒度更细的锁

    List<String> generate();

}

4 实现的aop

package com.xzx.statistics.redis.lock;

import com.google.common.collect.Lists;

import org.aspectj.lang.ProceedingJoinPoint;

import org.aspectj.lang.annotation.Around;

import org.aspectj.lang.annotation.Aspect;

import org.aspectj.lang.annotation.Pointcut;

import org.aspectj.lang.reflect.MethodSignature;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.core.annotation.Order;

import org.springframework.stereotype.Component;

import java.lang.reflect.Method;

import java.util.Arrays;

import java.util.List;

import java.util.Objects;

import java.util.stream.Collectors;

/**

 * 锁切面

 *

 * @author XuZhangxing

 */

@Aspect

@Component

@Order(1)

public final class LockAspect {

    private final LockService lockService;

    @Autowired(required = false)

    public LockAspect(LockService lockService) {

        this.lockService = lockService;

    }

    // 锁的目标对象

    @Pointcut("@annotation(com.xzx.statistics.redis.lock.LockMask)")

    public void lockTargetMethod() {

    }

    @Around("lockTargetMethod()")

    public Object doLock(ProceedingJoinPoint joinPoint) throws Throwable {

        // 获取到方法签名

        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();

        // 获取方法目标方法名称

        Method targetMethod = methodSignature.getMethod();

        // 获取实现类方法

        Method targetImplMethod =

                joinPoint.getTarget().getClass().getMethod(targetMethod.getName(), targetMethod.getParameterTypes());

        // 获取到注释对象

        LockMask lockMask = targetImplMethod.getAnnotation(LockMask.class);

        // 获取到目标对象

        Object targetObj = joinPoint.getArgs()[lockMask.index()];

        if (targetObj == null) {

            return joinPoint.proceed();

        }

        // 这里处理两种情况 一种基本数据类型的数据 二实现了KeyGenerate的对象，其他的都不处理

        List<LockParam> lockParamList = getLockKeyList(targetObj, lockMask);

        if (lockParamList.size() < 1) {

            return joinPoint.proceed();

        }

        // 加上分布式锁

        lockService.lock(lockParamList);

        try {

            // 正常业务逻辑 这里可以考虑启动一个异步线程来给当前的锁续期

            // 我觉得这样处理不太合理，不如直接将锁超时时间延长一些，有可能这个方法执行一直阻塞，锁无法释放

            return joinPoint.proceed();

        } finally {

            // 释放锁

            lockService.unLock(lockParamList);

        }

    }

    // 抽取锁的key

    private List<LockParam> getLockKeyList(Object target, LockMask lockMask) {

        String el = lockMask.el();

        if (el.trim().length() > 0) {

            // 适配el 表达式 暂时不支持

            throw new RuntimeException("function not supposed");

        }

        return getLockParams(target, lockMask);

    }

    // 递归遍历锁key

    private List<LockParam> getLockParams(Object target, LockMask lockMask) {

        LockKeyPrefix lockKeyPrefix = lockMask.lockKeyPre();

        long expireSecond = lockMask.expireSecond();

        if (target == null || target.getClass().isPrimitive()

                || target.getClass().isAssignableFrom(String.class)) {

            return Lists.newArrayList(new LockParam(lockKeyPrefix + ":" + target, expireSecond));

        }

        if (target instanceof KeyGenerate) {

            KeyGenerate keyGenerate = (KeyGenerate) target;

            // 注意对key去重

            List<String> keyList = keyGenerate.generate().stream().distinct().collect(Collectors.toList());

            return keyList.stream().map(key -> new LockParam(lockKeyPrefix + ":" + key, expireSecond))

                    .collect(Collectors.toList());

        }

        List<LockParam> result = Lists.newArrayList();

        if (target.getClass().isArray()) {

            Object[] arrayObj = (Object[]) target;

            Arrays.stream(arrayObj).filter(Objects::nonNull)

                    .forEach(obj -> result.addAll(getLockParams(obj, lockMask)));

        } else if (Iterable.class.isAssignableFrom(target.getClass())) {

            Iterable<?> iterable = (Iterable<?>) target;

            iterable.forEach(it -> result.addAll(getLockParams(it, lockMask)));

        }

        return result;

    }

}

总结：

1 锁的实现方式有多种，比如可以使用数据库、redis、zookeeper 等

2 入参是list<String> 这种类型或list<基本数据类型包装类>，也能适配，可以直接对整个list 加锁

3 最佳实践：建议将方法入参定义成对象，该对象实现 KeyGenerate 接口，这样复杂的入参也能由自己掌握key的生成规则，缺点是侵入性强，需要修改请求入参的dto

巴特西

Java 分布式锁实现的一些实践

最新文章

热门文章