分布式锁竟然能这么轻松实现？

最编程 2024-07-28 19:28:14

...

本文只讨论如何基于已实现了分布式锁的第三方框架进行二次封装，减少分布式锁的使用成本，而且当需要替换分布式锁实现时，只需要少量代码的调整，比如只需修改配置文件即可完成改造。
另外，本文是对另一篇文章中的实现的优化版，但主要思想是一致的。见使用Redisson实现分布式锁，Spring AOP简化之。

源码

在开始之前，我们先回想一个比较经典的场景——超卖，而解决超卖的一个方案就是加锁，真正扣减库存之前，必须拿到对应的锁，下面来看一段示例代码，其中锁的实现是借助了 Redisson：

@Transactional(rollbackFor = Throwable.class)
public void seckill(Long itemId, int purchaseCount) {
    RLock lock = redissonClient.getLock("item:" + itemId);
    boolean locked = false;

    try {
        locked = lock.tryLock(5000, TimeUnit.MILLISECONDS);
        if (locked) {
            doSeckill(itemId, purchaseCount);
        }

        
    } catch (InterruptedException e) {
        throw new RuntimeException("无法在指定时间内获得锁");
    } finally {
        // 是否未获得锁
        if (!locked) {
            throw new RuntimeException("尝试获取锁超时, 锁获取失败");
        }

        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        } else {
            throw new RuntimeException("锁释放失败, 当前线程不是锁的持有者");
        }
    }
}

public void doSeckill(Long itemId, int purchaseCount) {
    // 获取库存
    // 比较并扣减库存
    // 更新库存
    // 异步执行其他逻辑
}

上面的代码，看着也算还好，不太复杂，但其实除了真正的业务逻辑 doSeckill 外，其他的都是一些可模板化的代码结构。想象一下如果每一处使用分布式锁的地方都要写这堆东西，那还得了，更可怕的是，如果要换一种分布式锁的实现方式呢？

既然我们能预想到这种写法有这么严重的弊端，那就得想办法优化。最好能够实现：只需要关注真正的业务逻辑，需要使用分布式锁时，只需增加少量代码即可，比如加个注解；另外，如果需要更换分布式锁的实现方式，也不需要改任何代码。怎么实现呢？这就是这篇文章要解决的问题了。通过优化后，上面的示例代码将可以变得如下这么简单：

@DistributedLock(lockName = "#{#itemId}", lockNamePre = "item")
public void doSeckill(Long itemId, int purchaseCount) {
    // 获取库存
    // 比较并扣减库存
    // 更新库存
    // 异步执行其他逻辑
}

下面，正式开始~~~

Redisson概述

可参考另一篇文章的 Redisson概述。

使用Redisson实现分布式锁

1. 定义回调接口

/**
 * 分布锁回调接口
 */
public interface DistributedLockCallback<T> {

    /**
     * 调用者必须在此方法中实现需要加分布式锁的业务逻辑
     *
     * @return
     */
    public T process() throws Throwable;

    /**
     * 得到分布式锁名称
     *
     * @return
     */
    public String getLockName();

}

定义分布式具体实现的锁模板接口

/**
 * 分布式锁具体实现的模板接口
 *
 * @author sprainkle
 * @date 2019.04.20
 */
public interface DistributedLockTemplate {

    /** 尝试获取锁的默认等待时间 */
    long DEFAULT_WAIT_TIME = DistributedLock.DEFAULT_WAIT_TIME;
    /** 锁的默认超时时间. 超时后, 锁会被自动释放 */
    long DEFAULT_TIMEOUT = DistributedLock.DEFAULT_WAIT_TIME;
    /** 时间单位。默认为毫秒。 */
    TimeUnit DEFAULT_TIME_UNIT = DistributedLock.DEFAULT_TIME_UNIT;
    /** 获得锁名时拼接前后缀用到的分隔符 */
    String DEFAULT_SEPARATOR = DistributedLock.DEFAULT_SEPARATOR;
    /** lockName后缀 */
    String LOCK = DistributedLock.LOCK;

    /**
     * 使用分布式锁，使用锁默认超时时间。
     *
     * @param callback
     * @param fairLock 是否使用公平锁
     * @return
     */
    <T> T lock(DistributedLockCallback<T> callback, boolean fairLock);

    /**
     * 使用分布式锁。自定义锁的超时时间
     *
     * @param callback
     * @param leaseTime 锁超时时间。超时后自动释放锁。
     * @param timeUnit
     * @param fairLock  是否使用公平锁
     * @return
     */
    <T> T lock(DistributedLockCallback<T> callback, long leaseTime, TimeUnit timeUnit, boolean fairLock);

    /**
     * 尝试分布式锁，使用锁默认等待时间、超时时间。
     *
     * @param callback
     * @param <T>
     * @param fairLock 是否使用公平锁
     * @return
     */
    <T> T tryLock(DistributedLockCallback<T> callback, boolean fairLock);

    /**
     * 尝试分布式锁，自定义等待时间、超时时间。
     *
     * @param callback
     * @param waitTime  获取锁最长等待时间
     * @param leaseTime 锁超时时间。超时后自动释放锁。
     * @param timeUnit
     * @param <T>
     * @param fairLock  是否使用公平锁
     * @return
     */
    <T> T tryLock(DistributedLockCallback<T> callback, long waitTime, long leaseTime, TimeUnit timeUnit, boolean fairLock);

    /**
     * 锁是否由当前线程持有
     *
     * @param lock
     * @return
     */
    boolean isHeldByCurrentThread(Object lock);

}

基于 Redisson 实现

/**
 * Base DistributedLockTemplate. 用于封装一些公共方法
 */
public abstract class AbstractDistributedLockTemplate implements DistributedLockTemplate {
    /**
     * 处理业务逻辑
     *
     * @param callback
     * @param <T>
     * @return 业务逻辑处理结果
     */
    protected <T> T process(DistributedLockCallback<T> callback) {
        try {
            return callback.process();
        } catch (Throwable e) {
            if (e instanceof BaseException) {
                throw (BaseException) e;
            }

            throw new BaseException(CommonResponseEnum.SERVER_ERROR, null, e.getMessage(), e);
        }
    }
}

@Slf4j
public class RedisDistributedLockTemplate extends AbstractDistributedLockTemplate {

    private final RedissonClient redisson;
    private final String namespace;

    private final long lockTimeoutMs;
    private final long waitTimeoutMs;

    // 锁前缀
    private final String lockPrefix;
    // 锁后缀
    private final String lockPostfix;

    public RedisDistributedLockTemplate(RedissonClient redisson, DistributedLockProperties properties) {
        this.redisson = redisson;
        this.namespace = properties.getNamespace();
        this.lockTimeoutMs = Optional.ofNullable(properties.getLockTimeoutMs()).orElse(DEFAULT_TIMEOUT);
        this.waitTimeoutMs = Optional.ofNullable(properties.getWaitTimeoutMs()).orElse(DEFAULT_WAIT_TIME);

        this.lockPrefix = namespace + DEFAULT_SEPARATOR;
        this.lockPostfix = ".lock";
    }

    @Override
    public <T> T lock(DistributedLockCallback<T> callback, boolean fairLock) {
        return lock(callback, lockTimeoutMs, DEFAULT_TIME_UNIT, fairLock);
    }

    @Override
    public <T> T lock(DistributedLockCallback<T> callback, long leaseTime, TimeUnit timeUnit, boolean fairLock) {
        RLock lock = getLock(callback.getLockName(), fairLock);

        try {
            lock.lock(leaseTime, timeUnit);
            return process(callback);
        } finally {
            if (lock != null && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    @Override
    public <T> T tryLock(DistributedLockCallback<T> callback, boolean fairLock) {
        return tryLock(callback, waitTimeoutMs, lockTimeoutMs, DEFAULT_TIME_UNIT, fairLock);
    }

    @Override
    public <T> T tryLock(DistributedLockCallback<T> callback,
                         long waitTime,
                         long leaseTime,
                         TimeUnit timeUnit,
                         boolean fairLock) {
        RLock lock = getLock(callback.getLockName(), fairLock);
        boolean locked = false;

        DistributedLockContext context = DistributedLockContextHolder.getContext();
        context.setLock(lock);

        try {
            locked = lock.tryLock(waitTime, leaseTime, timeUnit);
            if (locked) {
                return process(callback);
            }
        } catch (InterruptedException e) {
            ResponseEnum.LOCK_NOT_YET_HOLD.assertFailWithMsg("无法在指定时间内获得锁", e);
        } finally {
            // 是否未获得锁
            if (!locked) {
                ResponseEnum.LOCK_NOT_YET_HOLD.assertFailWithMsg("尝试获取锁超时, 获取失败.");
            }

            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            } else {
                log.warn("锁释放失败, 当前线程不是锁的持有者");
            }
        }
        return null;
    }

    public RLock getLock(String lockName, boolean fairLock) {
        RLock lock;
        lockName = lockPrefix + lockName + lockPostfix;
        if (fairLock) {
            lock = redisson.getFairLock(lockName);
        } else {
            lock = redisson.getLock(lockName);
        }
        return lock;
    }

    @Override
    public boolean isHeldByCurrentThread(Object lock) {
        if (!(lock instanceof RLock)) {
            return false;
        }

        return ((RLock) lock).isHeldByCurrentThread();
    }
}

简单使用 RedisDistributedLockTemplate

DistributedLockTemplate lockTemplate = ...;
final String lockName = ...; 
lockTemplate.lock(new DistributedLockCallback<Object>() {
    @Override
    public Object process() {
        //do some business
        return null;
    }

    @Override
    public String getLockName() {
        return lockName;
    }
}, false);

会不会还是很麻烦？

虽说比使用原生 Redisson 时简单一点点，但是每次使用分布式锁都要写类似上面的重复代码，还是不够优雅。有没有什么方法可以只关注核心业务逻辑代码的编写，即上面的"do some business"。下面介绍如何使用Spring AOP来实现这一目标。

使用 Spring AOP 进一步封装

定义注解 DistributedLock

/**
 * <pre>
 *     可以使用该注解实现分布式锁。
 *
 *     获取lockName的优先级为：lockName > argNum > param
 *
 *     使用的是公平锁, 即先来先得.
 * </pre>
 */
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface DistributedLock {
    /**
     * 锁的名称。
     * 如果lockName可以确定，直接设置该属性。
     * <br><br>
     * 支持 SpEL, 格式为: #{expression}, 内置 #root, 属性包括: target, method, args 等, 其中 target 为注解所在类的 Spring Bean
     * 也支持 占位符 ${}
     */
    String lockName() default "";

    /**
     * lockName 前缀
     */
    String lockNamePre() default "";

    /**
     * lockName 后缀
     * @see #LOCK
     */
    String lockNamePost() default "";

    /**
     * 在开始加锁前, 执行某个方法进行校验.
     * <br><br>
     * 支持 SpEL, 格式为: #{expression}, 所以方法必须为 public, 如果方法的所在 Spring Bean 与注解的方法相同,
     * 写法为: #{#root.target.yourMethod(#param1, #param2, ...)}
     *
     * @return
     */
    String checkBefore() default "";

    /**
     * 获得锁名时拼接前后缀用到的分隔符
     * @see #DEFAULT_SEPARATOR
     */
    String separator() default DEFAULT_SEPARATOR;

    /**
     * <pre>
     *     获取注解的方法参数列表的某个参数对象的某个属性值来作为lockName。因为有时候lockName是不固定的。
     *     当param不为空时，可以通过argSeq参数来设置具体是参数列表的第几个参数，不设置则默认取第一个。
     * </pre>
     */
    String param() default "";

    /**
     * 将方法第argSeq个参数作为锁名. 0为无效值.
     */
    int argSeq() default 0;

    /**
     * 是否使用公平锁。
     * 公平锁即先来先得。
     */
    boolean fairLock() default false;

    /**
     * 是否使用尝试锁。
     */
    boolean tryLock() default true;

    /**
     * 最长等待时间。
     * 该字段只有当tryLock()返回true才有效。
     * @see #DEFAULT_WAIT_TIME
     */
    long waitTime() default DEFAULT_WAIT_TIME;

    /**
     * <pre>
     *     锁超时时间。
     *     超时时间过后，锁自动释放。
     *     建议：
     *       尽量缩简需要加锁的逻辑。
     * </pre>
     * @see #DEFAULT_TIMEOUT
     */
    long leaseTime() default DEFAULT_TIMEOUT;

    /**
     * 时间单位。默认为毫秒。
     */
    TimeUnit timeUnit() default TimeUnit.MILLISECONDS;

    /**
     * 尝试获取锁的默认等待时间
     */
    long DEFAULT_WAIT_TIME = 10000L;
    /**
     * 锁的默认超时时间. 超时后, 锁会被自动释放
     */
    long DEFAULT_TIMEOUT = 5000L;
    /**
     * 时间单位。默认为毫秒。
     */
    TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
    /**
     * 获得锁名时拼接前后缀用到的分隔符
     */
    String DEFAULT_SEPARATOR = ":";
    /**
     * lock
     */
    String LOCK = "lock";
}

定义切面织入的代码

/**
 * 分布式锁切面逻辑
 */
@Slf4j
@Aspect
public class DistributedLockAspect implements ApplicationContextAware, BeanFactoryAware, Ordered {

    /**
     * 解析模板
     */
    private static final ParserContext PARSER_CONTEXT = ParserContext.TEMPLATE_EXPRESSION;
    /**
     * SpEL 解析器
     */
    private static final SpelExpressionParser spElParser = new SpelExpressionParser();

    private ApplicationContext applicationContext;

    private BeanFactory beanFactory;
    /**
     * 用于解析 @BeanName 为对应的 Spring Bean
     */
    private BeanResolver beanResolver;

    private final DistributedLockTemplate lockTemplate;

    public DistributedLockAspect(DistributedLockTemplate lockTemplate) {
        this.lockTemplate = lockTemplate;
    }


    @Around(value = "@annotation(distributedLock)")
    public Object doAround(ProceedingJoinPoint pjp, DistributedLock distributedLock) {
        EvaluationContext evaluationCtx = getEvaluationContext(pjp);

        doCheckBefore(distributedLock, evaluationCtx);

        String lockName = getLockName(pjp, evaluationCtx);

        return lock(pjp, lockName);
    }

    /**
     * 执行 {@link DistributedLock#checkBefore()} 指定的方法
     *
     * @param distributedLock
     * @param evaluationCtx
     */
    private void doCheckBefore(DistributedLock distributedLock, EvaluationContext evaluationCtx) {
        String checkBefore = distributedLock.checkBefore();
        resolveExpression(evaluationCtx, checkBefore);
    }

    /**
     * 获取锁名
     *
     * @param jp
     * @return
     */
    private String getLockName(JoinPoint jp, EvaluationContext evaluationCtx) {
        DistributedLock annotation = getAnnotation(jp);
        String lockName = annotation.lockName();

        if (StrUtil.isNotBlank(lockName)) {
            lockName = resolveExpression(evaluationCtx, lockName);
        } else {
            Object[] args = jp.getArgs();
            if (args.length > 0) {
                String param = annotation.param();
                if (StrUtil.isNotBlank(param)) {
                    Object arg;
                    if (annotation.argSeq() > 0) {
                        arg = args[annotation.argSeq() - 1];
                    } else {
                        arg = args[0];
                    }
                    lockName = String.valueOf(getParam(arg, param));
                } else if (annotation.argSeq() > 0) {
                    lockName = String.valueOf(args[annotation.argSeq() - 1]);
                }
            }
        }

        if (StrUtil.isBlank(lockName)) {
            CommonResponseEnum.SERVER_ERROR.assertFailWithMsg("无法生成分布式锁锁名. annotation: {0}", annotation);
        }

        String preLockName = annotation.lockNamePre();
        String postLockName = annotation.lockNamePost();
        String separator = annotation.separator();

        if (StrUtil.isNotBlank(preLockName)) {
            lockName = preLockName + separator + lockName;
        }
        if (StrUtil.isNotBlank(postLockName)) {
            lockName = lockName + separator + postLockName;
        }

        return lockName;
    }

    /**
     * 从方法参数获取数据
     *
     * @param param
     * @param arg 方法的参数数组
     * @return
     */
    private Object getParam(Object arg, String param) {

        if (StrUtil.isNotBlank(param) && arg != null) {
            try {
                return BeanUtil.getFieldValue(arg, param);
            } catch (Exception e) {
                CommonResponseEnum.SERVER_ERROR.assertFailWithMsg("[{0}] 没有属性 [{1}]", arg.getClass(), param);
            }
        }
        return "";
    }

    /**
     * 获取锁并执行
     *
     * @param pjp
     * @param lockName
     * @return
     */
    private Object lock(ProceedingJoinPoint pjp, final String lockName) {
        DistributedLock annotation = PointCutUtils.getAnnotation(pjp, DistributedLock.class);

        boolean fairLock = annotation.fairLock();
        boolean tryLock = annotation.tryLock();

        if (tryLock) {
            return tryLock(pjp, annotation, lockName, fairLock);
        } else {
            return lock(pjp,lockName, fairLock);
        }
    }

    /**
     *
     * @param pjp
     * @param lockName
     * @param fairLock
     * @return
     */
    private Object lock(ProceedingJoinPoint pjp, final String lockName, boolean fairLock) {
        return lockTemplate.lock(new DistributedLockCallback<Object>() {
            @Override
            public Object process() throws Throwable {
                return pjp.proceed();
            }

            @Override
            public String getLockName() {
                return lockName;
            }
        }, fairLock);
    }

    /**
     *
     * @param pjp
     * @param annotation
     * @param lockName
     * @param fairLock
     * @return
     */
    private Object tryLock(ProceedingJoinPoint pjp, DistributedLock annotation, final String lockName, boolean fairLock) {

        long waitTime = annotation.waitTime(), leaseTime = annotation.leaseTime();
        TimeUnit timeUnit = annotation.timeUnit();

        return lockTemplate.tryLock(new DistributedLockCallback<Object>() {
            @Override
            public Object process() throws Throwable {
                return pjp.proceed();
            }

            @Override
            public String getLockName() {
                return lockName;
            }
        }, waitTime, leaseTime, timeUnit, fairLock);
    }

    // 省略若干
}

这里由于篇幅过长，只贴了主要代码，就不贴其他相关类的代码，如果有需要的可以去 Github 自行获取。

如何使用

如果是在本地测试，什么都不用配置，因为使用了 springboot-starter 的规范封装的，只需像其他 springboot-starter 一样，引入对应的依赖即可，类似如下。

<dependency>
    <groupId>com.sprainkle</groupId>
    <artifactId>spring-cloud-advance-common-lock</artifactId>
    <version>${spring-cloud-advance.version}</version>
</dependency>

然后，就可以像如下代码一样，直接在业务代码前加上分布式锁注解即可使用：

@DistributedLock(lockName = "#{#itemId}", lockNamePre = "item")
public void doSeckill(Long itemId, int purchaseCount) {
    // 获取库存
    // 比较并扣减库存
    // 更新库存
    // 异步执行其他逻辑
}

注解 DistributedLock 个参数的使用，可参考各参数的说明。

开始使用

因为这里直接在模块中编写测试用例，所以不用引入依赖。可参考源码

定义实体类 TestItem

@Data
@TableName("test_item")
public class TestItem {

    @TableId
    private Integer id;
    private String name;
    private Integer stock;

    public TestItem() {
    }

    public TestItem(Integer id, String name) {
        this.id = id;
        this.name = name;
    }
}

服务实现类 TestItemService

@Slf4j
@Service
public class TestItemService extends ServiceImpl<TestItemMapper, TestItem> {
    // 具体逻辑见下文
}

本次测试 ORM 框架使用了 Mybatis Plus，其他类，如 TestItemMapper 等，由于篇幅过长，这里就不展示了。

测试用例

定义 Worker 类

public static class Worker implements Runnable {

        private final CountDownLatch startSignal;
        private final CountDownLatch doneSignal;
        private final Action action;

        public Worker(CountDownLatch startSignal, CountDownLatch doneSignal, Action action) {
            this.startSignal = startSignal;
            this.doneSignal = doneSignal;
            this.action = action;
        }

        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName() + " start");
                // 阻塞, 直到接收到启动信号. 保证所有线程的起跑线是一样的, 即都是同时启动
                startSignal.await();
                // 具体逻辑
                action.execute();
                // 发送 已完成 信号
                doneSignal.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

该类是下文所有测试用例会用到的类。

testPlainLockName

首先我们定义 TestItemService 类，其中 initStock 为初始化库存，之后每个测试用例的第一步逻辑都是调用该方法，初始化/重置库存。

而 testPlainLockName 才是该测试用例的主要逻辑，逻辑也很简单，就不废话了，重点看分布式锁注解。可以看到要开启分布式锁，只需在方法签名加上 @DistributedLock(lockName = "item:1", lockNamePre = "item") 即可，非常方便。

那这么写是什么意思呢？其实就是，有很多线程在扣减商品(id=1)的库存前，需要拿到一把锁，锁名为 distributed-lock-test:item:1.lock，其中 distributed-lock-test:、.lock 是框架自己补进去的，剩下的 item:1 则是根据 lockName 和 lockNamePre 拼接的，拼接符默认为 :。

很明显，lockName 写死为 "1" 肯定不合适，这里只是演示需要，具体优化请继续往下，下文会给出。

@Slf4j
@Service
public class TestItemService extends ServiceImpl<TestItemMapper, TestItem> {

    private static final AtomicInteger i = new AtomicInteger(10);

    @Transactional(rollbackFor = Throwable.class)
    public TestItem initStock(Long id, Integer stock) {
        TestItem item = this.getById(id);

        if (item == null) {
            item = new TestItem(1, "牛奶");
        }

        item.setStock(stock);
        this.saveOrUpdate(item);

        return this.getById(id);
    }

    /**
     * 锁名为固定的字符串
     */
    @DistributedLock(lockName = "1", lockNamePre = "item")
    public Integer testPlainLockName(TestItem testItem) {
        TestItem item = this.getById(testItem.getId());
        Integer stock = item.getStock();

        if (stock > 0) {
            stock = stock - 1;
            item.setStock(stock);
            this.saveOrUpdate(item);
        } else {
            stock = -1;
        }

        return stock;
    }
}

接下来，定义测试用例类。

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = DistributedLockTestApplication.class)
public class DistributedLockTests {
    // 10 个线程一起跑
    private static int count = 10;

    @Autowired
    private TestItemService testItemService;

    @Test
    public void testPlainLockName() {
        Consumer<TestItem> consumer = testItem -> {
            Integer stock = testItemService.testPlainLockName(testItem);
            if (stock >= 0) {
                System.out.println(Thread.currentThread().getName() + ": rest stock = " + stock);
            } else {
                System.out.println(Thread.currentThread().getName() + ": sold out.");
            }
        };

        commonTest(consumer);
    }

    private void commonTest(Consumer<TestItem> consumer)  {
        try {
            CountDownLatch startSignal = new CountDownLatch(1);
            CountDownLatch doneSignal = new CountDownLatch(count);

            TestItem item = testItemService.initStock(1L, 8);

            for (int i = 0; i < count; ++i) {
                Action action = () -> consumer.accept(item);
                new Thread(new Worker(startSignal, doneSignal, action)).start();
            }
            
            // let all threads proceed
            startSignal.countDown(); 
            doneSignal.await();
            System.out.println("All processors done. Shutdown connection");
        } catch (Exception e) {
            log.error("", e);
        }
    }
}

其中，每次启动的线程数为 10，然后，commonTest(Consumer<TestItem> consumer) 为这里及之后所有测试用例的公共代码，所以抽象出来了，而 Consumer<TestItem> consumer 才是测试用例间不同的逻辑，并且每次都会把库存初始化为 8。

这里省略了各种配置文件和配置类等。

下文的所有截图，有一处单词拼写错误，rest stock 写成了 reset stock，还请将就着看。

启动测试用例，可以看到类似如下的控制台打印：

testPlainLockName

理论上应该只有8个线程能正常扣减库存，而结果也与预想的一样。这时如果去掉注解 @DistributedLock(lockName = "1", lockNamePre = "item")，会出现什么结果呢？类似如下：

without DistributedLock annotation

testSpel

很明显，上一个测试用例中，lockName 写死为 "1" 是不可取，而是应该取入参 testItem 的 id 的值，接下来，使用 SpEL 来实现该需求。

public class TestItemService {
    @DistributedLock(
            lockName = "#{#testItem.id}",
            lockNamePre = "item"
    )
    public Integer testSpel(TestItem testItem) {
        TestItem item = this.getById(testItem.getId());
        Integer stock = item.getStock();

        if (stock > 0) {
            stock = stock - 1;
            item.setStock(stock);
            this.saveOrUpdate(item);
        } else {
            stock = -1;
        }

        return stock;
    }
}

可以看到表达式 #{#testItem.id} 可能与大家以前使用的不太一样，这是因为在解析 SpEL 表达式时，使用了 解析模板 #{}，即表达式必须使用 #{} 包裹起来。其中对 SpEL 的支持，因为不是本文的重点，可以参考: https://cloud.tencent.com/developer/article/1497676。

public class DistributedLockTests {
    @Test
    public void testSpel() {
        Consumer<TestItem> consumer = testItem -> {
            Integer stock = testItemService.testSpel(testItem);
            if (stock >= 0) {
                System.out.println(Thread.currentThread().getName() + ": rest stock = " + stock);
            } else {
                System.out.println(Thread.currentThread().getName() + ": sold out.");
            }
        };

        commonTest(consumer);
    }
}

启动测试用例，可以看到控制台类似输出：

testSpel

testCheckBefore

public class TestItemService
    @DistributedLock(
            lockName = "#{#testItem.id}",
            lockNamePre = "item",
            checkBefore = "#{#root.target.check(#testItem)}"
    )
    public Integer testCheckBefore(TestItem testItem) {
        TestItem item = this.getById(testItem.getId());
        Integer stock = item.getStock();

        if (stock > 0) {
            stock = stock - 1;
            item.setStock(stock);
            this.saveOrUpdate(item);
        } else {
            stock = -1;
        }

        return stock;
    }

    public void check(TestItem testItem) {
        int randomInt = RandomUtil.randomInt(100, 10000);
        if (randomInt % 3 == 0) {
            System.out.println(String.format("current thread: %s, randomInt: %d", getCurrentThreadName(), randomInt));
            CommonResponseEnum.SERVER_BUSY.assertFail();
        }
    }
}

其中，参数 checkBefore 用于在开始加锁前, 执行某个方法进行校验，比如这里，使用方法 check 模拟了流量控制，符合一定条件时，直接抛异常返回。

public class DistributedLockTests {
    @Test
    public void testCheckBefore() {
        Consumer<TestItem> consumer = testItem -> {

            Integer stock = -1;

            try {
                stock = testItemService.testCheckBefore(testItem);
            } catch (Exception e) {
                System.out.println(Thread.currentThread().getName() + ": 系统繁忙");
                return;
            }

            if (stock >= 0) {
                System.out.println(Thread.currentThread().getName() + ": rest stock = " + stock);
            } else {
                System.out.println(Thread.currentThread().getName() + ": sold out.");
            }
        };

        commonTest(consumer);
    }
}

启动测试用例，类似控制台输出如下：

testCheckBefore

这里有一点需要注意，方法 check 必须为 public。

testTryLock、testFairLock、testWaitTime

这几个测试用例比较简单，这里就不展示了，有兴趣的可以自己跑一下。

testLeaseTime

首先来理解一下参数 leaseTime 的作用，即：锁超时时间，超时时间过后，锁自动释放。

挺好理解的，但重点是，锁被自动释放后，之前执行的逻辑需要怎么处理。
在最后释放锁的时候，发现锁已经不是当前线程持有，有可能已经被其他持有，那之前获得锁后执行的逻辑，都变得不可信了，所以理论上需要撤销，如果是数据库操作，那就是回滚。

首先来看 testLeaseTime 的第一个测试用例

public class TestItemService
    @DistributedLock(
            lockName = "#{#testItem.id}",
            lockNamePre = "item",
            leaseTime = 2000
    )
    public Integer testLeaseTime(TestItem testItem) {
        int ci = TestItemService.i.getAndDecrement();
        if (ci == 10) {
            sleep(5000L);
            log.info("模拟阻塞完成");
        } else {
            sleep(300L);
        }

        TestItem item = this.getById(testItem.getId());
        Integer stock = item.getStock();

        if (stock > 0) {
            stock = stock - 1;
            item.setStock(stock);
            this.saveOrUpdate(item);
        } else {
            stock = -1;
        }

        return stock;
    }

    private void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class DistributedLockTests {
    @Test
    public void testLeaseTime() {
        Consumer<TestItem> consumer = testItem -> {

            Integer stock = testItemService.testLeaseTime(testItem);

            if (stock >= 0) {
                System.out.println(Thread.currentThread().getName() + ": rest stock = " + stock);
            } else {
                System.out.println(Thread.currentThread().getName() + ": sold out.");
            }
        };

        commonTest(consumer);
    }
}

启动测试用例，结果类似如下：

testLeaseTime

这样乍一看，看下没什么毛病，数据库的库存也是正常的，为 0。但再细看一下代码，模拟阻塞是在一开始就进行的，那如果把该部分代码挪到从数据库获取到数据之后，会发生什么呢？

来看第二个测试用例：

public class TestItemService
    /**
     * 超卖
     *
     * @param testItem
     * @return
     */
    @DistributedLock(
            lockName = "#{#testItem.id}",
            lockNamePre = "item",
            leaseTime = 2000
    )
    public Integer testLeaseTimeOversold(TestItem testItem) {
        TestItem item = this.getById(testItem.getId());

        int ci = TestItemService.i.getAndDecrement();
        if (ci == 10) {
            sleep(5000L);
            log.info("模拟阻塞完成");
        } else {
            sleep(300L);
        }

        Integer stock = item.getStock();

        if (stock > 0) {
            stock = stock - 1;
            item.setStock(stock);
            this.saveOrUpdate(item);
        } else {
            stock = -1;
        }

        return stock;
    }
}

public class DistributedLockTests {
    @Test
    public void testLeaseTimeOversold() {
        Consumer<TestItem> consumer = testItem -> {

            Integer stock = testItemService.testLeaseTimeOversold(testItem);

            if (stock >= 0) {
                System.out.println(Thread.currentThread().getName() + ": rest stock = " + stock);
            } else {
                System.out.println(Thread.currentThread().getName() + ": sold out.");
            }
        };

        commonTest(consumer);
    }
}

启动测试用例，结果类似如下：

testLeaseTimeOversold

可以看到，很明显超卖了，sold out 之后，居然还能继续扣减库存，这时候数据库的库存应该也是 7，为什么呢？
因为它扣减库存时，锁已经被释放了，未持有锁的情况下，扣减的库存大概率都是有问题的，只要并发足够大。

testLeaseTimeWithTransactional

public class TestItemService
    @DistributedLock(
            lockName = "#{#testItem.id}",
            lockNamePre = "item",
            leaseTime = 2000
    )
    @Transactional(rollbackFor = Throwable.class)
    public Integer testLeaseTimeWithTransactional(TestItem testItem) {
        TestItem item = this.getById(testItem.getId());
        Integer stock = item.getStock();

        int ci = TestItemService.i.getAndDecrement();
        if (ci == 10) {
            sleep(5000L);
            log.info("模拟阻塞完成");
        } else {
            sleep(300L);
        }

        if (stock > 0) {
            stock = stock - 1;
            item.setStock(stock);
            this.saveOrUpdate(item);
        } else {
            stock = -1;
        }

        return stock;
    }
}

public class DistributedLockTests {
    @Test
    public void testLeaseTimeWithTransactional() {
        Consumer<TestItem> consumer = testItem -> {

            Integer stock = testItemService.testLeaseTimeWithTransactional(testItem);

            if (stock >= 0) {
                System.out.println(Thread.currentThread().getName() + ": rest stock = " + stock);
            } else {
                System.out.println(Thread.currentThread().getName() + ": sold out.");
            }
        };

        commonTest(consumer);
    }
}

启动测试用例，结果类似如下：

testLeaseTimeWithTransactional

可以看到，虽然只有一个线程打印 "sold out", 但 "Thread-11" 也扣减失败了，所以超卖并没有出现。这时的库存为 0。
上图中，输出的东西比其他其他测试用例多了异常的堆栈日志，在哪里抛出来的呢？如下：

2021-08-16 16:23:32.332  INFO 96595 --- [      Thread-11] c.s.s.c.a.c.l.service.TestItemService    : 模拟阻塞完成
2021-08-16 16:23:32.349  WARN 96595 --- [      Thread-11] c.a.c.l.a.i.RedisDistributedLockTemplate : 锁释放失败, 当前线程不是锁的持有者
com.sprainkle.spring.cloud.advance.common.core.exception.BusinessException: 系统繁忙，请稍后重试
    at com.sprainkle.spring.cloud.advance.common.core.exception.assertion.BusinessExceptionAssert.newException(BusinessExceptionAssert.java:35)
    at com.sprainkle.spring.cloud.advance.common.core.exception.assertion.Assert.newExceptionWithMsg(Assert.java:52)
    at com.sprainkle.spring.cloud.advance.common.core.exception.assertion.Assert.assertFailWithMsg(Assert.java:789)
    at com.sprainkle.spring.cloud.advance.common.lock.api.UnlockFailureProcessor.beforeCommit(UnlockFailureProcessor.java:28)
    at org.springframework.transaction.support.TransactionSynchronizationUtils.triggerBeforeCommit(TransactionSynchronizationUtils.java:96)
    // ... 省略若干
    at com.sprainkle.spring.cloud.advance.common.lock.DistributedLockTests.lambda$commonTest$8(DistributedLockTests.java:185)
    at com.sprainkle.spring.cloud.advance.common.lock.DistributedLockTests$Worker.run(DistributedLockTests.java:217)
    at java.lang.Thread.run(Thread.java:748)
Caused by: com.sprainkle.spring.cloud.advance.common.core.exception.WrapMessageException: 释放锁时, 当前线程不是锁的持有者
    at com.sprainkle.spring.cloud.advance.common.core.exception.assertion.Assert.newExceptionWithMsg(Assert.java:51)
    ... 33 more

这里可以抛出一个问题，为什么加上 @Transactional(rollbackFor = Throwable.class) 注解后，就不会出现超卖呢？仅仅是因为 @Transactional 注解，还是还有其他原因？

与该项目相关的第一行日志为：at *.lock.api.UnlockFailureProcessor.beforeCommit(UnlockFailureProcessor.java:28)，且最后的 Caused by 为 释放锁时, 当前线程不是锁的持有者，大概可以猜出：数据库的交互在最后的 commit 前，判断了当前线程是否为锁的持有者，如果不是，则抛异常让数据回滚。

这里可以简单给出 UnlockFailureProcessor 的源码：

/**
 * 锁释放失败时的处理器. 如果当前线程不是锁的持有者, 直接抛异常让数据回滚
 */
public class UnlockFailureProcessor implements TransactionSynchronization {

    private final Object lock;

    private final DistributedLockTemplate distributedLockTemplate;

    public UnlockFailureProcessor(DistributedLockTemplate distributedLockTemplate, Object lock) {
        this.distributedLockTemplate = distributedLockTemplate;
        this.lock = lock;
    }

    @Override
    public void beforeCommit(boolean readOnly) {
        boolean heldByCurrentThread = distributedLockTemplate.isHeldByCurrentThread(lock);

        if (!heldByCurrentThread) {
            ResponseEnum.LOCK_NO_MORE_HOLD.assertFailWithMsg("释放锁时, 当前线程不是锁的持有者");
        }
    }

}

因此，可以得出结论，其实真正起作用的并不仅仅是因为加了 @Transactional 注解，还需要有相应的其他支持，@Transactional 注解只是让其拥有管理事务的环境，方便数据回滚。

如果有兴趣，可以将 ResponseEnum.LOCK_NO_MORE_HOLD.assertFailWithMsg("释放锁时, 当前线程不是锁的持有者"); 注释，然后再跑一遍，结果为：

testLeaseTimeWithoutRollback

不出意外的话，这时的库存为 7。

基于 ZooKeeper 实现

基于 ZooKeeper 的分布式锁实现，在源码中已给出，请参考实现类 ZooDistributedLockTemplate。使用的时候，只需将配置调整为：

sca-common:
  distributed:
    lock:
      impl: zoo
      zoo:
        # zookeeper服务器地址. 多个时用','分开
        connectString: "127.0.0.1:2181"
        # zookeeper的session过期时间. 即锁的过期时间. 可用于全局配置锁的过期时间
        sessionTimeoutMs: 10000
        # zookeeper的连接超时时间
        connectionTimeoutMs: 15000

当然也是需要引入相关依赖：

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
        </dependency>

结语

至此，本文的主要内容已介绍完毕，由于本文的篇幅过长，贴了太多代码，所以只演示了简单使用，当业务比较复杂的时候，上面的使用方法可能没办法很好的支持；另外，在使用过程中也有需要注意的地方，不然有可能出现分布式锁注解不生效的情况；还有部分关键代码的原理以及背后的原因，都不好在这里一一说明，只能放到另外一篇文章做详细分析。

然后，这篇文章仅作为抛砖引玉，如果有其他更好的方案，欢迎留言，一起讨论学习。
谢谢！！！

上一篇：如何正确运用Redisson锁的tryLock功能

下一篇： FileLock

分布式锁竟然能这么轻松实现？

Redisson概述

使用Redisson实现分布式锁

1. 定义回调接口

定义分布式具体实现的锁模板接口

基于 Redisson 实现

简单使用 RedisDistributedLockTemplate

会不会还是很麻烦？

使用 Spring AOP 进一步封装

定义注解 DistributedLock

定义切面织入的代码

如何使用

开始使用

定义实体类 TestItem

服务实现类 TestItemService

测试用例

定义 Worker 类

testPlainLockName

testSpel

testCheckBefore

testTryLock、testFairLock、testWaitTime

testLeaseTime

testLeaseTimeWithTransactional

基于 ZooKeeper 实现

结语

分布式锁竟然能这么轻松实现？

Springboot与Redisson整合教程：轻松实现分布式锁之七

原来查询也能这么简单！轻松实现你的查询需求！