diff --git a/easyflow-commons/easyflow-common-cache/src/main/java/tech/easyflow/common/cache/RedisLockExecutor.java b/easyflow-commons/easyflow-common-cache/src/main/java/tech/easyflow/common/cache/RedisLockExecutor.java index 17d040bc..94c12718 100644 --- a/easyflow-commons/easyflow-common-cache/src/main/java/tech/easyflow/common/cache/RedisLockExecutor.java +++ b/easyflow-commons/easyflow-common-cache/src/main/java/tech/easyflow/common/cache/RedisLockExecutor.java @@ -1,5 +1,6 @@ package tech.easyflow.common.cache; +import jakarta.annotation.PreDestroy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -10,6 +11,11 @@ import org.springframework.stereotype.Component; import java.time.Duration; import java.util.Collections; import java.util.UUID; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Supplier; /** @@ -27,6 +33,13 @@ public class RedisLockExecutor { private static final DefaultRedisScript NEXT_FENCING_TOKEN_SCRIPT; private static final DefaultRedisScript ACQUIRE_FENCED_LOCK_SCRIPT; + private final ScheduledExecutorService lockRenewalExecutor = + Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "easyflow-redis-lock-renewal"); + thread.setDaemon(true); + return thread; + }); + static { RELEASE_LOCK_SCRIPT = new DefaultRedisScript<>(); RELEASE_LOCK_SCRIPT.setScriptText( @@ -94,6 +107,66 @@ public class RedisLockExecutor { } } + /** + * 在自动续租的分布式锁保护下执行任务。 + * + *

适用于包含数据库锁等待或外部持久化操作、无法由固定租约严格覆盖的管理命令。 + * 若执行期间确认锁已丢失，则不向调用方返回成功。

+ */ + public void executeWithRenewingLock( + String lockKey, + Duration waitTimeout, + Duration leaseTimeout, + Runnable task) { + executeWithRenewingLock(lockKey, waitTimeout, leaseTimeout, () -> { + task.run(); + return null; + }); + } + + /** + * 在自动续租的分布式锁保护下执行有返回值任务。 + */ + public T executeWithRenewingLock( + String lockKey, + Duration waitTimeout, + Duration leaseTimeout, + Supplier task) { + LockHandle handle = acquire(lockKey, waitTimeout, leaseTimeout); + AtomicBoolean lost = new AtomicBoolean(); + long renewalIntervalMillis = Math.max(1L, leaseTimeout.toMillis() / 3L); + ScheduledFuture renewal = lockRenewalExecutor.scheduleWithFixedDelay( + () -> { + try { + if (!handle.renew()) { + lost.set(true); + } + } catch (RuntimeException exception) { + lost.set(true); + log.warn("分布式锁续租失败，当前命令不得返回成功: lockKey={}", + lockKey, exception); + } + }, + renewalIntervalMillis, + renewalIntervalMillis, + TimeUnit.MILLISECONDS); + try { + T result = task.get(); + if (lost.get()) { + throw new IllegalStateException("执行期间分布式锁已丢失，lockKey=" + lockKey); + } + return result; + } finally { + renewal.cancel(false); + handle.release(); + } + } + + @PreDestroy + public void shutdownLockRenewalExecutor() { + lockRenewalExecutor.shutdownNow(); + } + /** * 获取显式释放的分布式锁句柄。 * diff --git a/easyflow-commons/easyflow-common-cache/src/test/java/tech/easyflow/common/cache/RedisLockExecutorTest.java b/easyflow-commons/easyflow-common-cache/src/test/java/tech/easyflow/common/cache/RedisLockExecutorTest.java index 3f286a53..73fb5b45 100644 --- a/easyflow-commons/easyflow-common-cache/src/test/java/tech/easyflow/common/cache/RedisLockExecutorTest.java +++ b/easyflow-commons/easyflow-common-cache/src/test/java/tech/easyflow/common/cache/RedisLockExecutorTest.java @@ -11,6 +11,9 @@ import org.springframework.data.redis.core.script.RedisScript; import java.lang.reflect.Field; import java.time.Duration; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; /** * {@link RedisLockExecutor} 回归测试。 @@ -147,6 +150,96 @@ public class RedisLockExecutorTest { String.valueOf(Duration.ofDays(4).toMillis()))); } + @Test + public void renewingLockShouldRenewBeforeLongRunningCommandCompletes() throws Exception { + StringRedisTemplate redisTemplate = Mockito.mock(StringRedisTemplate.class); + ValueOperations valueOperations = mockValueOperations(true); + CountDownLatch renewed = new CountDownLatch(1); + Mockito.when(redisTemplate.opsForValue()).thenReturn(valueOperations); + Mockito.when(redisTemplate.execute( + ArgumentMatchers.>any(), + ArgumentMatchers.>any(), + ArgumentMatchers.anyString(), + ArgumentMatchers.anyString() + )).thenAnswer(invocation -> { + renewed.countDown(); + return 1L; + }); + + RedisLockExecutor executor = new RedisLockExecutor(); + setRedisTemplate(executor, redisTemplate); + try { + executor.executeWithRenewingLock( + "easyflow:test:renewing-lock", + Duration.ZERO, + Duration.ofMillis(60), + () -> { + try { + Assert.assertTrue(renewed.await(1, TimeUnit.SECONDS)); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new AssertionError("等待锁续租时被中断", exception); + } + }); + } finally { + executor.shutdownLockRenewalExecutor(); + } + + Mockito.verify(redisTemplate, Mockito.atLeastOnce()).execute( + ArgumentMatchers.>any(), + ArgumentMatchers.eq(List.of("easyflow:test:renewing-lock")), + ArgumentMatchers.anyString(), + ArgumentMatchers.eq("60")); + } + + @Test + public void renewingLockMustNotReturnSuccessAfterRenewalThrows() throws Exception { + StringRedisTemplate redisTemplate = Mockito.mock(StringRedisTemplate.class); + ValueOperations valueOperations = mockValueOperations(true); + CountDownLatch renewalAttempted = new CountDownLatch(1); + AtomicInteger scriptCalls = new AtomicInteger(); + Mockito.when(redisTemplate.opsForValue()).thenReturn(valueOperations); + Mockito.when(redisTemplate.execute( + ArgumentMatchers.>any(), + ArgumentMatchers.>any(), + ArgumentMatchers.anyString(), + ArgumentMatchers.anyString() + )).thenAnswer(invocation -> { + if (scriptCalls.incrementAndGet() == 1) { + renewalAttempted.countDown(); + throw new IllegalStateException("redis unavailable"); + } + return 1L; + }); + + RedisLockExecutor executor = new RedisLockExecutor(); + setRedisTemplate(executor, redisTemplate); + try { + try { + executor.executeWithRenewingLock( + "easyflow:test:renewal-failure", + Duration.ZERO, + Duration.ofMillis(60), + () -> { + try { + Assert.assertTrue(renewalAttempted.await(1, TimeUnit.SECONDS)); + // 等待续租线程把失败结果发布到调用线程；业务任务与续租 + // 同时完成时，锁仍处于原租约内且 callback 已结束。 + Thread.sleep(50L); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new AssertionError(exception); + } + }); + Assert.fail("续租异常后不应返回成功"); + } catch (IllegalStateException exception) { + Assert.assertTrue(exception.getMessage().contains("分布式锁已丢失")); + } + } finally { + executor.shutdownLockRenewalExecutor(); + } + } + @SuppressWarnings("unchecked") private ValueOperations mockValueOperations(boolean acquired) { ValueOperations valueOperations = Mockito.mock(ValueOperations.class);