fix(cache): 击穿等待响应中断 + interval 兜底,消除空转与无限循环
未抢到锁的自旋等待原以 waited += waitIntervalMs 推进:waitIntervalMs<=0 时永不 前进 → 无限循环 hang;且 sleep 吞中断后不退出,被中断时以 Redis 往返为节奏空转 到超时、不响应协作式取消。改为溢出安全的 deadline 截止判断 + Math.max(1,interval) 兜底,sleep 返回中断信号、被中断即 break 降级并保留中断标志。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -185,11 +185,13 @@ public class RedisCacheEnhanceTemplate {
|
||||
}
|
||||
}
|
||||
|
||||
// 未抢到锁:自旋等待重读,超时降级直接回源
|
||||
long waited = 0;
|
||||
while (waited < bd.getWaitTimeoutMs()) {
|
||||
sleep(bd.getWaitIntervalMs());
|
||||
waited += bd.getWaitIntervalMs();
|
||||
// 未抢到锁:自旋等待重读,超时或被中断则降级直接回源
|
||||
long intervalMs = Math.max(1, bd.getWaitIntervalMs()); // 兜底:interval<=0 时不前进会空转
|
||||
long deadlineNanos = System.nanoTime() + bd.getWaitTimeoutMs() * 1_000_000L;
|
||||
while (System.nanoTime() - deadlineNanos < 0) { // 溢出安全的截止判断
|
||||
if (!sleep(intervalMs)) {
|
||||
break; // 被中断:停止等待、降级回源(中断标志已保留,响应协作式取消)
|
||||
}
|
||||
Lookup<T> r3 = tryGet(dataKey, decoder);
|
||||
if (r3.hit) {
|
||||
if (r3.realValue) {
|
||||
@@ -198,7 +200,7 @@ public class RedisCacheEnhanceTemplate {
|
||||
return r3.value;
|
||||
}
|
||||
}
|
||||
log.warn("[redis-cache] breakdown wait timeout, fallback to loader, key={}", key);
|
||||
log.warn("[redis-cache] breakdown wait timeout/interrupted, fallback to loader, key={}", key);
|
||||
return loader.get();
|
||||
}
|
||||
|
||||
@@ -240,11 +242,14 @@ public class RedisCacheEnhanceTemplate {
|
||||
}
|
||||
}
|
||||
|
||||
private static void sleep(long ms) {
|
||||
/** 休眠;返回 false 表示被中断(已重置中断标志,调用方应停止等待)。 */
|
||||
private static boolean sleep(long ms) {
|
||||
try {
|
||||
Thread.sleep(ms);
|
||||
return true;
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -24,12 +24,14 @@ import java.util.function.Supplier;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyList;
|
||||
import static org.mockito.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.atMost;
|
||||
import static org.mockito.Mockito.lenient;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
@@ -264,4 +266,32 @@ class RedisCacheEnhanceTemplateTest {
|
||||
assertThrows(IllegalArgumentException.class,
|
||||
() -> template.getOrLoad("user:1", User.class, 0, () -> new User(1L, "a")));
|
||||
}
|
||||
|
||||
// ---- #3+#4: 击穿等待的中断响应与 interval 兜底 ----
|
||||
|
||||
@Test
|
||||
void getOrLoadWaitIntervalZeroDoesNotSpinForever() {
|
||||
props.getBreakdown().setWaitIntervalMs(0); // 误配 0:旧实现 waited 永不前进 → 无限循环
|
||||
props.getBreakdown().setWaitTimeoutMs(50);
|
||||
when(valueOps.get("cache:user:1")).thenReturn(null);
|
||||
when(valueOps.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(false);
|
||||
User got = assertTimeoutPreemptively(Duration.ofSeconds(2), () ->
|
||||
template.getOrLoad("user:1", User.class, 60, () -> new User(7L, "fb")));
|
||||
assertEquals("fb", got.getName());
|
||||
}
|
||||
|
||||
@Test
|
||||
void getOrLoadInterruptedDuringWaitStopsSpinning() {
|
||||
props.getBreakdown().setWaitIntervalMs(20);
|
||||
props.getBreakdown().setWaitTimeoutMs(10000); // 大超时:旧实现中断后空转数百次到超时
|
||||
when(valueOps.get("cache:user:1")).thenReturn(null);
|
||||
when(valueOps.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(false);
|
||||
Thread.currentThread().interrupt(); // 当前线程预置中断,getOrLoad 同步执行可感知
|
||||
User got = template.getOrLoad("user:1", User.class, 60, () -> new User(7L, "fb"));
|
||||
boolean wasInterrupted = Thread.interrupted(); // 读取并清除,避免污染后续测试
|
||||
assertEquals("fb", got.getName());
|
||||
assertTrue(wasInterrupted, "中断标志应被保留(协作式取消)");
|
||||
// 中断后立即停止等待:仅首次 tryGet 读一次,而非空转数百次
|
||||
verify(valueOps, atMost(2)).get("cache:user:1");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user