fix: 补齐分块同步所需的锁续租能力
- 定向回补 develop 已有的 Redis 自动续租锁及回归测试
This commit is contained in:
@@ -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<Long> NEXT_FENCING_TOKEN_SCRIPT;
|
||||
private static final DefaultRedisScript<Long> 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 {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 在自动续租的分布式锁保护下执行任务。
|
||||
*
|
||||
* <p>适用于包含数据库锁等待或外部持久化操作、无法由固定租约严格覆盖的管理命令。
|
||||
* 若执行期间确认锁已丢失，则不向调用方返回成功。</p>
|
||||
*/
|
||||
public void executeWithRenewingLock(
|
||||
String lockKey,
|
||||
Duration waitTimeout,
|
||||
Duration leaseTimeout,
|
||||
Runnable task) {
|
||||
executeWithRenewingLock(lockKey, waitTimeout, leaseTimeout, () -> {
|
||||
task.run();
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 在自动续租的分布式锁保护下执行有返回值任务。
|
||||
*/
|
||||
public <T> T executeWithRenewingLock(
|
||||
String lockKey,
|
||||
Duration waitTimeout,
|
||||
Duration leaseTimeout,
|
||||
Supplier<T> 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();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取显式释放的分布式锁句柄。
|
||||
*
|
||||
|
||||
@@ -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<String, String> valueOperations = mockValueOperations(true);
|
||||
CountDownLatch renewed = new CountDownLatch(1);
|
||||
Mockito.when(redisTemplate.opsForValue()).thenReturn(valueOperations);
|
||||
Mockito.when(redisTemplate.execute(
|
||||
ArgumentMatchers.<RedisScript<Long>>any(),
|
||||
ArgumentMatchers.<List<String>>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.<RedisScript<Long>>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<String, String> valueOperations = mockValueOperations(true);
|
||||
CountDownLatch renewalAttempted = new CountDownLatch(1);
|
||||
AtomicInteger scriptCalls = new AtomicInteger();
|
||||
Mockito.when(redisTemplate.opsForValue()).thenReturn(valueOperations);
|
||||
Mockito.when(redisTemplate.execute(
|
||||
ArgumentMatchers.<RedisScript<Long>>any(),
|
||||
ArgumentMatchers.<List<String>>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<String, String> mockValueOperations(boolean acquired) {
|
||||
ValueOperations<String, String> valueOperations = Mockito.mock(ValueOperations.class);
|
||||
|
||||
Reference in New Issue
Block a user