|
|
@@ -1,621 +0,0 @@
|
|
|
-package com.bex.staking.engine;
|
|
|
-
|
|
|
-import static org.assertj.core.api.Assertions.assertThat;
|
|
|
-import static org.mockito.ArgumentMatchers.anyString;
|
|
|
-import static org.mockito.Mockito.mock;
|
|
|
-import static org.mockito.Mockito.when;
|
|
|
-
|
|
|
-import com.bex.staking.annotation.Idempotent;
|
|
|
-import com.bex.staking.constant.StakingErrorCode;
|
|
|
-import com.bex.staking.exception.StakingException;
|
|
|
-import java.lang.reflect.InvocationHandler;
|
|
|
-import java.lang.reflect.Proxy;
|
|
|
-import java.util.Map;
|
|
|
-import java.util.concurrent.ConcurrentHashMap;
|
|
|
-import java.util.concurrent.ConcurrentLinkedQueue;
|
|
|
-import java.util.concurrent.CountDownLatch;
|
|
|
-import java.util.concurrent.ExecutorService;
|
|
|
-import java.util.concurrent.Executors;
|
|
|
-import java.util.concurrent.Future;
|
|
|
-import java.util.concurrent.TimeUnit;
|
|
|
-import java.util.concurrent.atomic.AtomicInteger;
|
|
|
-import java.util.concurrent.locks.Condition;
|
|
|
-import java.util.concurrent.locks.ReentrantLock;
|
|
|
-import org.aspectj.lang.annotation.Aspect;
|
|
|
-import org.junit.jupiter.api.AfterEach;
|
|
|
-import org.junit.jupiter.api.BeforeEach;
|
|
|
-import org.junit.jupiter.api.DisplayName;
|
|
|
-import org.junit.jupiter.api.Test;
|
|
|
-import org.redisson.api.RLock;
|
|
|
-import org.redisson.api.RedissonClient;
|
|
|
-import org.springframework.aop.aspectj.annotation.AspectJProxyFactory;
|
|
|
-
|
|
|
-/**
|
|
|
- * Redisson 分布式锁并发集成测试(任务 17.4 Redisson 锁部分)。
|
|
|
- *
|
|
|
- * <p>被测组件:{@link IdempotencyBarrier}({@link Aspect} 切面,基于 Redisson {@link RLock} 的
|
|
|
- * {@code tryLock(waitTime, leaseTime, unit)})。本测试通过 Spring AOP({@link AspectJProxyFactory})将
|
|
|
- * <strong>真实的</strong> {@code IdempotencyBarrier} 切面织入一个标注 {@link Idempotent} 的样例 Bean,对其方法
|
|
|
- * 发起多线程并发调用,验证设计文档"测试策略 - Redisson:验证锁获取/释放与 leaseTime 自动释放"所述语义:</p>
|
|
|
- *
|
|
|
- * <ol>
|
|
|
- * <li><b>并发互斥与重复请求拒绝(需求 14.1、14.2)</b>:相同幂等键的并发请求中,有且仅有一个获取锁并执行业务,
|
|
|
- * 其余因 {@code waitTime=0} 锁被占用立即抛出 {@code StakingException(40008)};</li>
|
|
|
- * <li><b>不同键互不阻塞</b>:不同幂等键的并发请求各自获取独立锁,全部成功执行;</li>
|
|
|
- * <li><b>业务完成释放锁(需求 14.3)</b>:同一键串行调用,前一次完成后释放锁,后一次可再次获取并执行;</li>
|
|
|
- * <li><b>leaseTime 到期自动释放(需求 14.4)</b>:持锁线程未主动释放时,超过 {@code leaseTime} 后锁可被重新获取
|
|
|
- * (分别在锁层面与切面层面各验证一次)。</li>
|
|
|
- * </ol>
|
|
|
- *
|
|
|
- * <h2>环境说明(方案退化)</h2>
|
|
|
- *
|
|
|
- * <p>本运行 / CI 环境<strong>无 Docker、无可用 Redis、本地仓库亦无嵌入式 Redis 依赖</strong>
|
|
|
- * ({@code embedded-redis} 等),无法启动真实 {@code RedissonClient}。为不引入不可用依赖而阻塞,本测试采用
|
|
|
- * <strong>内存版 RLock 语义模拟</strong>({@link SimulatedLock})替代真实 Redis:它忠实复现 Redisson
|
|
|
- * {@code RLock} 的关键并发契约——按键互斥、可重入、{@code tryLock(waitTime, leaseTime)} 的等待/占用语义、以及
|
|
|
- * <strong>显式 leaseTime 到期自动释放</strong>(不含 watchdog 续约,因切面始终显式传入 leaseTime)。被测对象始终是
|
|
|
- * 生产代码 {@code IdempotencyBarrier},仅"锁后端"被替换为内存模拟,故对锁获取/释放、40008 拒绝、leaseTime
|
|
|
- * 自动释放等行为的验证结论可迁移到真实 Redis 部署。锁键拼接、SpEL 求值等切面逻辑则完全走真实实现。</p>
|
|
|
- *
|
|
|
- * <p>若后续环境具备 Docker 或可用 Redis,可将 {@link #redissonClient} 替换为真实
|
|
|
- * {@code Redisson.create(config)},本测试用例与断言可直接复用。</p>
|
|
|
- *
|
|
|
- * <p>Validates: Requirements 14.3, 14.4(并附带覆盖 14.1、14.2)
|
|
|
- */
|
|
|
-@DisplayName("Redisson 分布式锁并发集成测试 (内存 RLock 语义模拟驱动真实切面)")
|
|
|
-class RedissonLockConcurrencyTest {
|
|
|
-
|
|
|
- /** 模拟的 RedissonClient:getLock 返回按键唯一、由 {@link SimulatedLock} 支撑的 RLock 代理。 */
|
|
|
- private RedissonClient redissonClient;
|
|
|
-
|
|
|
- /** 按锁键缓存的 RLock 代理实例(保证同一键多线程拿到同一把锁)。 */
|
|
|
- private Map<String, RLock> rlockByKey;
|
|
|
-
|
|
|
- /** 织入真实 IdempotencyBarrier 切面后的样例服务代理。 */
|
|
|
- private SampleIdempotentService proxy;
|
|
|
-
|
|
|
- /** 样例服务目标对象(持有执行计数与同步闩)。 */
|
|
|
- private SampleIdempotentService target;
|
|
|
-
|
|
|
- @BeforeEach
|
|
|
- void setUp() {
|
|
|
- rlockByKey = new ConcurrentHashMap<>();
|
|
|
- redissonClient = mock(RedissonClient.class);
|
|
|
- // getLock 在并发下被多线程调用:以 ConcurrentHashMap.computeIfAbsent 保证按键唯一、线程安全
|
|
|
- when(redissonClient.getLock(anyString())).thenAnswer(inv -> {
|
|
|
- String key = inv.getArgument(0);
|
|
|
- return rlockByKey.computeIfAbsent(key, RedissonLockConcurrencyTest::newSimulatedRLock);
|
|
|
- });
|
|
|
-
|
|
|
- // 用 Spring AOP 将真实 IdempotencyBarrier 切面织入样例 Bean(CGLIB 代理具体类)
|
|
|
- IdempotencyBarrier barrier = new IdempotencyBarrier(redissonClient);
|
|
|
- target = new SampleIdempotentService();
|
|
|
- AspectJProxyFactory factory = new AspectJProxyFactory(target);
|
|
|
- factory.addAspect(barrier);
|
|
|
- factory.setProxyTargetClass(true);
|
|
|
- proxy = factory.getProxy();
|
|
|
- }
|
|
|
-
|
|
|
- @AfterEach
|
|
|
- void tearDown() {
|
|
|
- // 释放可能仍阻塞在 proceedLatch 上的样例方法,避免线程泄漏
|
|
|
- target.releaseAll();
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 1. 并发互斥:相同键仅一个执行,其余 40008(需求 14.1、14.2)
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- @Test
|
|
|
- @DisplayName("相同幂等键的 6 个并发请求仅一个获取锁执行,其余 5 个被拒绝返回 40008")
|
|
|
- void concurrentSameKey_onlyOneProceeds() throws Exception {
|
|
|
- int n = 6;
|
|
|
- String key = "user-1001:idem-A";
|
|
|
- target.useBlocking(); // 获胜线程进入业务后阻塞,保证其余线程在其持锁期间发起竞争
|
|
|
-
|
|
|
- ConcurrentLinkedQueue<Outcome> outcomes = new ConcurrentLinkedQueue<>();
|
|
|
- CountDownLatch startGate = new CountDownLatch(1);
|
|
|
- ExecutorService pool = Executors.newFixedThreadPool(n);
|
|
|
- try {
|
|
|
- for (int i = 0; i < n; i++) {
|
|
|
- pool.submit(() -> {
|
|
|
- try {
|
|
|
- startGate.await();
|
|
|
- String r = proxy.blocking(key);
|
|
|
- outcomes.add(Outcome.success(r));
|
|
|
- } catch (StakingException e) {
|
|
|
- outcomes.add(Outcome.rejected(e.getCode()));
|
|
|
- } catch (Throwable t) {
|
|
|
- outcomes.add(Outcome.error(t));
|
|
|
- }
|
|
|
- return null;
|
|
|
- });
|
|
|
- }
|
|
|
- startGate.countDown(); // 同时放行,制造最大竞争
|
|
|
-
|
|
|
- // 等待"获胜线程已进入业务"且"其余 5 个线程已全部失败返回"(获胜线程仍阻塞持锁)
|
|
|
- assertThat(target.awaitEntered(5, TimeUnit.SECONDS)).as("应有一个线程获取锁进入业务").isTrue();
|
|
|
- awaitUntil(() -> outcomes.size() >= n - 1, 5_000);
|
|
|
-
|
|
|
- // 此刻获胜线程仍持锁阻塞,落败的 5 个请求必然已被 40008 拒绝
|
|
|
- long rejected = outcomes.stream().filter(o -> o.type == Outcome.Type.REJECTED).count();
|
|
|
- assertThat(rejected).as("应有 n-1 个并发请求被幂等屏障拒绝").isEqualTo(n - 1);
|
|
|
-
|
|
|
- target.releaseAll(); // 放行获胜线程完成
|
|
|
- awaitUntil(() -> outcomes.size() == n, 5_000);
|
|
|
- } finally {
|
|
|
- pool.shutdownNow();
|
|
|
- }
|
|
|
-
|
|
|
- long success = outcomes.stream().filter(o -> o.type == Outcome.Type.SUCCESS).count();
|
|
|
- long rejected = outcomes.stream().filter(o -> o.type == Outcome.Type.REJECTED).count();
|
|
|
- long errored = outcomes.stream().filter(o -> o.type == Outcome.Type.ERROR).count();
|
|
|
-
|
|
|
- assertThat(success).as("有且仅有一个请求获取锁并执行业务").isEqualTo(1);
|
|
|
- assertThat(rejected).as("其余请求均被拒绝返回 40008").isEqualTo(n - 1);
|
|
|
- assertThat(errored).as("不应出现非预期异常").isZero();
|
|
|
- assertThat(target.executionCount()).as("业务方法应仅真正执行一次").isEqualTo(1);
|
|
|
- outcomes.stream()
|
|
|
- .filter(o -> o.type == Outcome.Type.REJECTED)
|
|
|
- .forEach(o -> assertThat(o.code).isEqualTo(StakingErrorCode.DUPLICATE_REQUEST));
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 2. 不同键互不阻塞:各自获取独立锁全部成功
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- @Test
|
|
|
- @DisplayName("不同幂等键的并发请求各自获取独立锁,全部成功执行、互不阻塞")
|
|
|
- void concurrentDifferentKeys_allProceed() throws Exception {
|
|
|
- int n = 8;
|
|
|
- ConcurrentLinkedQueue<Outcome> outcomes = new ConcurrentLinkedQueue<>();
|
|
|
- CountDownLatch startGate = new CountDownLatch(1);
|
|
|
- ExecutorService pool = Executors.newFixedThreadPool(n);
|
|
|
- try {
|
|
|
- for (int i = 0; i < n; i++) {
|
|
|
- final String key = "user-" + i + ":idem-" + i; // 互不相同的键
|
|
|
- pool.submit(() -> {
|
|
|
- try {
|
|
|
- startGate.await();
|
|
|
- outcomes.add(Outcome.success(proxy.fast(key)));
|
|
|
- } catch (StakingException e) {
|
|
|
- outcomes.add(Outcome.rejected(e.getCode()));
|
|
|
- } catch (Throwable t) {
|
|
|
- outcomes.add(Outcome.error(t));
|
|
|
- }
|
|
|
- return null;
|
|
|
- });
|
|
|
- }
|
|
|
- startGate.countDown();
|
|
|
- pool.shutdown();
|
|
|
- assertThat(pool.awaitTermination(15, TimeUnit.SECONDS)).isTrue();
|
|
|
- } finally {
|
|
|
- pool.shutdownNow();
|
|
|
- }
|
|
|
-
|
|
|
- long success = outcomes.stream().filter(o -> o.type == Outcome.Type.SUCCESS).count();
|
|
|
- assertThat(success).as("不同键不应互相阻塞,应全部成功").isEqualTo(n);
|
|
|
- assertThat(target.executionCount()).as("每个键各执行一次").isEqualTo(n);
|
|
|
- assertThat(rlockByKey).as("应为每个不同键各创建一把锁").hasSize(n);
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 3. 业务完成释放锁:同键可串行复用(需求 14.3)
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- @Test
|
|
|
- @DisplayName("同一幂等键串行调用:前一次完成释放锁后,后一次可再次获取并执行(需求 14.3)")
|
|
|
- void lockReleasedAfterCompletion_sequentialReuseSucceeds() {
|
|
|
- String key = "user-2002:idem-seq";
|
|
|
- // 三次串行调用同一键,若锁未在完成后释放,第二次起将被 40008 拒绝
|
|
|
- assertThat(proxy.fast(key)).isEqualTo("ok:" + key);
|
|
|
- assertThat(proxy.fast(key)).isEqualTo("ok:" + key);
|
|
|
- assertThat(proxy.fast(key)).isEqualTo("ok:" + key);
|
|
|
- assertThat(target.executionCount()).as("串行复用应每次都成功执行").isEqualTo(3);
|
|
|
-
|
|
|
- // 锁应已完全释放(无人持有)
|
|
|
- SimulatedLock lock = currentSimulatedLock("staking:idem:sample:fast:" + key);
|
|
|
- assertThat(lock).isNotNull();
|
|
|
- assertThat(lock.isLocked()).as("业务完成后锁应已释放").isFalse();
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 4a. leaseTime 自动释放(锁层面,需求 14.4)
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- @Test
|
|
|
- @DisplayName("锁层面:持锁线程不释放时,超过 leaseTime 后锁被自动释放、可被重新获取(需求 14.4)")
|
|
|
- void leaseTimeAutoRelease_atLockLevel() throws Exception {
|
|
|
- RLock lock = redissonClient.getLock("staking:idem:lease-test");
|
|
|
- long leaseMillis = 400L;
|
|
|
-
|
|
|
- // 线程 T1 获取锁后"崩溃"——永不主动释放
|
|
|
- CountDownLatch acquired = new CountDownLatch(1);
|
|
|
- Thread holder = new Thread(() -> {
|
|
|
- try {
|
|
|
- boolean ok = lock.tryLock(0, leaseMillis, TimeUnit.MILLISECONDS);
|
|
|
- if (ok) {
|
|
|
- acquired.countDown();
|
|
|
- }
|
|
|
- // 故意不调用 unlock(),模拟持锁进程异常
|
|
|
- } catch (InterruptedException e) {
|
|
|
- Thread.currentThread().interrupt();
|
|
|
- }
|
|
|
- });
|
|
|
- holder.start();
|
|
|
- assertThat(acquired.await(2, TimeUnit.SECONDS)).as("T1 应成功获取锁").isTrue();
|
|
|
-
|
|
|
- // 主线程立即尝试:锁被占用,应失败
|
|
|
- assertThat(lock.tryLock(0, leaseMillis, TimeUnit.MILLISECONDS))
|
|
|
- .as("leaseTime 未到期时锁应仍被占用,获取失败")
|
|
|
- .isFalse();
|
|
|
-
|
|
|
- // 等待超过 leaseTime 后再次尝试:锁应已自动释放,可重新获取
|
|
|
- Thread.sleep(leaseMillis + 300L);
|
|
|
- assertThat(lock.tryLock(0, leaseMillis, TimeUnit.MILLISECONDS))
|
|
|
- .as("超过 leaseTime 后锁应自动释放并可被重新获取,避免死锁")
|
|
|
- .isTrue();
|
|
|
- lock.unlock();
|
|
|
- holder.join(2_000);
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 4b. leaseTime 自动释放(切面层面,需求 14.4)
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- @Test
|
|
|
- @DisplayName("切面层面:慢业务持锁超过 leaseTime 后,同键的另一并发请求因锁自动释放得以执行(需求 14.4)")
|
|
|
- void leaseTimeAutoRelease_viaAspectAllowsReentryAfterExpiry() throws Exception {
|
|
|
- String key = "user-3003:idem-lease";
|
|
|
- target.useSlowShortLease(); // 业务进入后阻塞,且 @Idempotent leaseTime=1s
|
|
|
-
|
|
|
- ExecutorService pool = Executors.newFixedThreadPool(2);
|
|
|
- try {
|
|
|
- // 线程 A:获取锁、进入业务后阻塞(远超 1s 的 leaseTime)
|
|
|
- Future<Outcome> fa = pool.submit(() -> {
|
|
|
- try {
|
|
|
- return Outcome.success(proxy.slowShortLease(key));
|
|
|
- } catch (StakingException e) {
|
|
|
- return Outcome.rejected(e.getCode());
|
|
|
- } catch (Throwable t) {
|
|
|
- return Outcome.error(t);
|
|
|
- }
|
|
|
- });
|
|
|
- assertThat(target.awaitEntered(3, TimeUnit.SECONDS)).as("线程 A 应先获取锁进入业务").isTrue();
|
|
|
-
|
|
|
- // 等待超过 leaseTime(1s),使 A 的锁租约到期自动释放
|
|
|
- Thread.sleep(1_400L);
|
|
|
-
|
|
|
- // 线程 B:同键再次请求。若 leaseTime 未自动释放,将被 40008 拒绝;自动释放后应能获取锁执行
|
|
|
- Future<Outcome> fb = pool.submit(() -> {
|
|
|
- try {
|
|
|
- return Outcome.success(proxy.slowShortLease(key));
|
|
|
- } catch (StakingException e) {
|
|
|
- return Outcome.rejected(e.getCode());
|
|
|
- } catch (Throwable t) {
|
|
|
- return Outcome.error(t);
|
|
|
- }
|
|
|
- });
|
|
|
-
|
|
|
- // B 进入业务即说明它成功获取了锁(执行计数升至 2)
|
|
|
- awaitUntil(() -> target.executionCount() >= 2, 5_000);
|
|
|
- assertThat(target.executionCount())
|
|
|
- .as("A 的锁租约到期自动释放后,B 应成功获取锁并执行")
|
|
|
- .isEqualTo(2);
|
|
|
-
|
|
|
- target.releaseAll(); // 放行 A、B 完成
|
|
|
- Outcome oa = fa.get(5, TimeUnit.SECONDS);
|
|
|
- Outcome ob = fb.get(5, TimeUnit.SECONDS);
|
|
|
- assertThat(oa.type).as("线程 A 不应异常").isIn(Outcome.Type.SUCCESS, Outcome.Type.REJECTED);
|
|
|
- assertThat(ob.type).as("线程 B 应成功执行(锁已自动释放)").isEqualTo(Outcome.Type.SUCCESS);
|
|
|
- } finally {
|
|
|
- pool.shutdownNow();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 辅助:获取当前键对应的 SimulatedLock(用于断言锁状态)
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- private SimulatedLock currentSimulatedLock(String lockKey) {
|
|
|
- RLock rlock = rlockByKey.get(lockKey);
|
|
|
- return rlock == null ? null : SIM_BY_PROXY.get(rlock);
|
|
|
- }
|
|
|
-
|
|
|
- /** 轮询等待条件成立,最多等待 {@code timeoutMillis} 毫秒。 */
|
|
|
- private static void awaitUntil(java.util.function.BooleanSupplier condition, long timeoutMillis)
|
|
|
- throws InterruptedException {
|
|
|
- long deadline = System.currentTimeMillis() + timeoutMillis;
|
|
|
- while (System.currentTimeMillis() < deadline) {
|
|
|
- if (condition.getAsBoolean()) {
|
|
|
- return;
|
|
|
- }
|
|
|
- Thread.sleep(10L);
|
|
|
- }
|
|
|
- if (!condition.getAsBoolean()) {
|
|
|
- throw new AssertionError("等待条件超时(" + timeoutMillis + "ms)");
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 样例被测 Bean:标注 @Idempotent,由真实切面拦截
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- /**
|
|
|
- * 携带 {@link Idempotent} 注解的样例服务,模拟"确认质押 / 收益领取"等高频写接口。
|
|
|
- *
|
|
|
- * <p>提供三种业务方法:</p>
|
|
|
- *
|
|
|
- * <ul>
|
|
|
- * <li>{@link #fast(String)}:立即返回(leaseTime 30s,足够长不会在测试期内到期);</li>
|
|
|
- * <li>{@link #blocking(String)}:进入后阻塞在 {@code proceedLatch},用于制造并发互斥窗口;</li>
|
|
|
- * <li>{@link #slowShortLease(String)}:进入后阻塞,且 {@code leaseTime=1s},用于验证租约自动释放。</li>
|
|
|
- * </ul>
|
|
|
- */
|
|
|
- public static class SampleIdempotentService {
|
|
|
-
|
|
|
- private final AtomicInteger executions = new AtomicInteger();
|
|
|
- private volatile CountDownLatch enteredLatch = new CountDownLatch(1);
|
|
|
- private volatile CountDownLatch proceedLatch = new CountDownLatch(0);
|
|
|
- private volatile boolean blockMode = false;
|
|
|
-
|
|
|
- void useBlocking() {
|
|
|
- this.executions.set(0);
|
|
|
- this.enteredLatch = new CountDownLatch(1);
|
|
|
- this.proceedLatch = new CountDownLatch(1);
|
|
|
- this.blockMode = true;
|
|
|
- }
|
|
|
-
|
|
|
- void useSlowShortLease() {
|
|
|
- this.executions.set(0);
|
|
|
- this.enteredLatch = new CountDownLatch(1);
|
|
|
- this.proceedLatch = new CountDownLatch(1);
|
|
|
- this.blockMode = true;
|
|
|
- }
|
|
|
-
|
|
|
- int executionCount() {
|
|
|
- return executions.get();
|
|
|
- }
|
|
|
-
|
|
|
- boolean awaitEntered(long timeout, TimeUnit unit) throws InterruptedException {
|
|
|
- return enteredLatch.await(timeout, unit);
|
|
|
- }
|
|
|
-
|
|
|
- void releaseAll() {
|
|
|
- CountDownLatch latch = this.proceedLatch;
|
|
|
- while (latch.getCount() > 0) {
|
|
|
- latch.countDown();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /** 立即返回的幂等方法(leaseTime 30s,长于测试时长,确保仅靠主动释放回收锁)。 */
|
|
|
- @Idempotent(prefix = "sample:fast", key = "#key", leaseTime = 30L, unit = TimeUnit.SECONDS)
|
|
|
- public String fast(String key) {
|
|
|
- executions.incrementAndGet();
|
|
|
- return "ok:" + key;
|
|
|
- }
|
|
|
-
|
|
|
- /** 进入后阻塞的幂等方法(leaseTime 30s),用于并发互斥窗口。 */
|
|
|
- @Idempotent(prefix = "sample:blocking", key = "#key", leaseTime = 30L, unit = TimeUnit.SECONDS)
|
|
|
- public String blocking(String key) throws InterruptedException {
|
|
|
- executions.incrementAndGet();
|
|
|
- enteredLatch.countDown();
|
|
|
- if (blockMode) {
|
|
|
- proceedLatch.await(5, TimeUnit.SECONDS);
|
|
|
- }
|
|
|
- return "ok:" + key;
|
|
|
- }
|
|
|
-
|
|
|
- /** 进入后阻塞、leaseTime=1s 的幂等方法,用于验证 leaseTime 到期自动释放。 */
|
|
|
- @Idempotent(prefix = "sample:slow", key = "#key", leaseTime = 1L, unit = TimeUnit.SECONDS)
|
|
|
- public String slowShortLease(String key) throws InterruptedException {
|
|
|
- executions.incrementAndGet();
|
|
|
- enteredLatch.countDown();
|
|
|
- if (blockMode) {
|
|
|
- proceedLatch.await(5, TimeUnit.SECONDS);
|
|
|
- }
|
|
|
- return "ok:" + key;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 内存版 RLock 语义模拟
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- /** RLock 代理 → 其后端 SimulatedLock 的映射,便于测试断言锁内部状态。 */
|
|
|
- private static final Map<RLock, SimulatedLock> SIM_BY_PROXY = new ConcurrentHashMap<>();
|
|
|
-
|
|
|
- /**
|
|
|
- * 创建一把由 {@link SimulatedLock} 支撑的 {@link RLock} 动态代理。
|
|
|
- *
|
|
|
- * <p>仅实现 {@code IdempotencyBarrier} 实际调用的三个方法:{@code tryLock(long,long,TimeUnit)}、
|
|
|
- * {@code unlock()}、{@code isHeldByCurrentThread()}(外加 {@code isLocked()}/{@code getName()} 供断言),
|
|
|
- * 其余方法抛出 {@link UnsupportedOperationException},避免无意中依赖未模拟的 Redisson 行为。</p>
|
|
|
- */
|
|
|
- private static RLock newSimulatedRLock(String key) {
|
|
|
- SimulatedLock sim = new SimulatedLock();
|
|
|
- InvocationHandler handler = (proxyObj, method, args) -> {
|
|
|
- switch (method.getName()) {
|
|
|
- case "tryLock":
|
|
|
- if (args != null && args.length == 3) {
|
|
|
- return sim.tryLock((Long) args[0], (Long) args[1], (TimeUnit) args[2]);
|
|
|
- }
|
|
|
- throw new UnsupportedOperationException("未模拟的 tryLock 重载");
|
|
|
- case "unlock":
|
|
|
- if (args == null || args.length == 0) {
|
|
|
- sim.unlock();
|
|
|
- return null;
|
|
|
- }
|
|
|
- throw new UnsupportedOperationException("未模拟的 unlock 重载");
|
|
|
- case "isHeldByCurrentThread":
|
|
|
- return sim.isHeldByCurrentThread();
|
|
|
- case "isLocked":
|
|
|
- return sim.isLocked();
|
|
|
- case "getName":
|
|
|
- return key;
|
|
|
- case "toString":
|
|
|
- return "SimulatedRLock[" + key + "]";
|
|
|
- case "hashCode":
|
|
|
- return System.identityHashCode(proxyObj);
|
|
|
- case "equals":
|
|
|
- return proxyObj == args[0];
|
|
|
- default:
|
|
|
- throw new UnsupportedOperationException("SimulatedRLock 未模拟的方法: " + method.getName());
|
|
|
- }
|
|
|
- };
|
|
|
- RLock rlock = (RLock) Proxy.newProxyInstance(
|
|
|
- RedissonLockConcurrencyTest.class.getClassLoader(), new Class<?>[] {RLock.class}, handler);
|
|
|
- SIM_BY_PROXY.put(rlock, sim);
|
|
|
- return rlock;
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 忠实复现 Redisson {@link RLock} 关键并发契约的内存锁:
|
|
|
- *
|
|
|
- * <ul>
|
|
|
- * <li>按线程互斥、可重入;</li>
|
|
|
- * <li>{@code tryLock(waitTime, leaseTime, unit)}:锁空闲即获取并按 leaseTime 设定租约;被他人占用时在
|
|
|
- * waitTime 内等待,超时返回 {@code false};</li>
|
|
|
- * <li><b>leaseTime 到期自动释放</b>:持锁超过 leaseTime 后,下一次 {@code tryLock}/{@code isHeldByCurrentThread}
|
|
|
- * 检查时惰性回收(对应 Redisson 显式 leaseTime 到期释放、无 watchdog 续约的语义);</li>
|
|
|
- * <li>{@code unlock()}:仅持有者可释放,非持有者(如租约已到期被他人重获)调用视为无操作。</li>
|
|
|
- * </ul>
|
|
|
- */
|
|
|
- static final class SimulatedLock {
|
|
|
-
|
|
|
- private final ReentrantLock guard = new ReentrantLock();
|
|
|
- private final Condition freed = guard.newCondition();
|
|
|
- private Thread owner;
|
|
|
- private int holdCount;
|
|
|
- private long leaseExpireAtMillis;
|
|
|
-
|
|
|
- boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
|
|
|
- long waitNanos = unit.toNanos(Math.max(0L, waitTime));
|
|
|
- long leaseMillis = leaseTime > 0 ? unit.toMillis(leaseTime) : -1L;
|
|
|
- long deadline = System.nanoTime() + waitNanos;
|
|
|
- guard.lockInterruptibly();
|
|
|
- try {
|
|
|
- while (true) {
|
|
|
- expireIfNeeded();
|
|
|
- if (owner == null) {
|
|
|
- owner = Thread.currentThread();
|
|
|
- holdCount = 1;
|
|
|
- renewLease(leaseMillis);
|
|
|
- return true;
|
|
|
- }
|
|
|
- if (owner == Thread.currentThread()) {
|
|
|
- // 可重入
|
|
|
- holdCount++;
|
|
|
- renewLease(leaseMillis);
|
|
|
- return true;
|
|
|
- }
|
|
|
- long remaining = deadline - System.nanoTime();
|
|
|
- if (remaining <= 0) {
|
|
|
- return false;
|
|
|
- }
|
|
|
- // 限制单次等待时长,确保能及时醒来重新检查租约到期
|
|
|
- long sleepNanos = remaining;
|
|
|
- if (leaseExpireAtMillis != Long.MAX_VALUE) {
|
|
|
- long leaseLeft = leaseExpireAtMillis - System.currentTimeMillis();
|
|
|
- long leaseLeftNanos =
|
|
|
- leaseLeft > 0 ? TimeUnit.MILLISECONDS.toNanos(leaseLeft) : 1_000_000L;
|
|
|
- sleepNanos = Math.min(remaining, Math.max(leaseLeftNanos, 1_000_000L));
|
|
|
- }
|
|
|
- freed.awaitNanos(sleepNanos);
|
|
|
- }
|
|
|
- } finally {
|
|
|
- guard.unlock();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- void unlock() {
|
|
|
- guard.lock();
|
|
|
- try {
|
|
|
- if (owner != Thread.currentThread()) {
|
|
|
- // 租约已到期且被他人重新获取:当前线程已非持有者,按无操作处理(不抛异常以保证测试稳定)
|
|
|
- return;
|
|
|
- }
|
|
|
- holdCount--;
|
|
|
- if (holdCount <= 0) {
|
|
|
- release();
|
|
|
- }
|
|
|
- } finally {
|
|
|
- guard.unlock();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- boolean isHeldByCurrentThread() {
|
|
|
- guard.lock();
|
|
|
- try {
|
|
|
- expireIfNeeded();
|
|
|
- return owner == Thread.currentThread();
|
|
|
- } finally {
|
|
|
- guard.unlock();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- boolean isLocked() {
|
|
|
- guard.lock();
|
|
|
- try {
|
|
|
- expireIfNeeded();
|
|
|
- return owner != null;
|
|
|
- } finally {
|
|
|
- guard.unlock();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private void renewLease(long leaseMillis) {
|
|
|
- leaseExpireAtMillis = leaseMillis > 0 ? System.currentTimeMillis() + leaseMillis : Long.MAX_VALUE;
|
|
|
- }
|
|
|
-
|
|
|
- private void expireIfNeeded() {
|
|
|
- if (owner != null && System.currentTimeMillis() >= leaseExpireAtMillis) {
|
|
|
- release();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private void release() {
|
|
|
- owner = null;
|
|
|
- holdCount = 0;
|
|
|
- leaseExpireAtMillis = 0L;
|
|
|
- freed.signalAll();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // ========================================================================
|
|
|
- // 测试结果记录
|
|
|
- // ========================================================================
|
|
|
-
|
|
|
- /** 单次并发调用的结果归类。 */
|
|
|
- private static final class Outcome {
|
|
|
- enum Type {
|
|
|
- SUCCESS,
|
|
|
- REJECTED,
|
|
|
- ERROR
|
|
|
- }
|
|
|
-
|
|
|
- final Type type;
|
|
|
- final String value;
|
|
|
- final int code;
|
|
|
- final Throwable error;
|
|
|
-
|
|
|
- private Outcome(Type type, String value, int code, Throwable error) {
|
|
|
- this.type = type;
|
|
|
- this.value = value;
|
|
|
- this.code = code;
|
|
|
- this.error = error;
|
|
|
- }
|
|
|
-
|
|
|
- static Outcome success(String value) {
|
|
|
- return new Outcome(Type.SUCCESS, value, 0, null);
|
|
|
- }
|
|
|
-
|
|
|
- static Outcome rejected(int code) {
|
|
|
- return new Outcome(Type.REJECTED, null, code, null);
|
|
|
- }
|
|
|
-
|
|
|
- static Outcome error(Throwable t) {
|
|
|
- return new Outcome(Type.ERROR, null, -1, t);
|
|
|
- }
|
|
|
- }
|
|
|
-}
|