zhou-hao · Oct 15, 2026 · Oct 10, 2026 · Oct 10, 2026 · Oct 10, 2026 · Oct 10, 2026
diff --git a/.github/workflows/pull_request_5x.yml b/.github/workflows/pull_request_5x.yml
@@ -8,6 +8,16 @@ jobs:
   build:
 
     runs-on: ubuntu-latest
+    services:
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     steps:
     - uses: actions/checkout@v4
       with:

diff --git a/README.md b/README.md
@@ -128,6 +128,20 @@ public Mono<String> addAccount(@RequestBody Mono<Account> account){
 
 ```
 
+## 本地响应式缓存失效
+
+`hsweb-concurrent/hsweb-concurrent-cache` 中的 Guava、Caffeine 缓存在
+`evict`、`evictAll` 处理对应 key 时撤销已登记的加载者，`clear` 撤销处理时已有的加载者。
+加载成功时通过现有加载映射在同一 key 的原子操作内校验身份并同步回填。
+已撤销加载者的完成或错误不会回填缓存、移除之后的新加载者或清除其缓存值。
+
+getter 在订阅时才登记加载者并开始缓存读取，避免仅装配 Publisher 就捕获旧值。
+已经开始读取且与 `clear` 重叠的在途请求可以完成原值。`clear` 不是全缓存事务，
+不保证撤销并发产生的新订阅，也不改变调用方默认加载来源的新鲜度。
+
+Redis 的写入是异步远程操作，仍沿用原失效逻辑；上述本地保证不适用于
+Redis 同节点异步写入或跨节点写入与失效。
+
 ## License
 
 [Apache 2.0](https://github.com/spring-projects/spring-boot/blob/main/LICENSE.txt)

diff --git a/...rent-cache/src/main/java/org/hswebframework/web/cache/supports/AbstractReactiveCache.java b/...rent-cache/src/main/java/org/hswebframework/web/cache/supports/AbstractReactiveCache.java
@@ -16,8 +16,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
+/**
+ * 管理响应式缓存的在途加载与共享结果，存储操作由具体后端实现。
+ * 本地同步后端以加载者身份保护回填和失败清理，不提供异步远程后端的原子保证。
+ */
 @Slf4j
 public abstract class AbstractReactiveCache<E> implements ReactiveCache<E> {
     static final Sinks.EmitFailureHandler emitFailureHandler = Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(30));
@@ -58,9 +63,12 @@ private void tryLoad(ContextView context) {
                     source = source
                         .switchIfEmpty((Mono) defaultValue
                             .flatMap(val -> {
-                                return parent.putNow(key, val).thenReturn(val);
+                                return parent.putLoadedValue(this, val).thenReturn(val);
                             }));
                 }
+                // 在 complete 撤销身份前清理失败，避免旧加载失败误删新加载者或新值。
+                source = source.onErrorResume(err -> parent.handleLoaderError(
+                    key, err, parent.evictLoading(key, this)));
                 loading = source.subscribe(
                     val -> {
                         complete();
@@ -95,11 +103,50 @@ private void complete() {
 
     public abstract Mono<Void> putNow(Object key, Object value);
 
+    private Mono<Void> putLoadedValue(CacheLoader owner, Object value) {
+        AtomicReference<Mono<Void>> write = new AtomicReference<>(Mono.empty());
+        cacheLoading.computeIfPresent(owner.key, (key, current) -> {
+            if (current == owner) {
+                // 本地缓存同步写入，与同一 key 的失效共用加载映射的原子操作。
+                write.set(putNow(key, value));
+            }
+            return current;
+        });
+        return write.get();
+    }
+
+    protected final void invalidateLoading(Object key, Runnable invalidation) {
+        cacheLoading.compute(key, (_key, current) -> {
+            invalidation.run();
+            return null;
+        });
+    }
+
+    protected final void invalidateLoading(Object key, CacheLoader expected, Runnable invalidation) {
+        cacheLoading.computeIfPresent(key, (_key, current) -> {
+            if (current == expected) {
+                invalidation.run();
+                return null;
+            }
+            return current;
+        });
+    }
+
+    protected final void invalidateAllLoading(Runnable invalidation) {
+        // 撤销处理时已登记的加载者；与 clear 重叠的新订阅不属于全缓存事务。
+        cacheLoading.clear();
+        invalidation.run();
+    }
+
+    private CacheLoader createLoader(Object key) {
+        // 读取也延迟到订阅，不能在装配时捕获失效前的本地值。
+        return new CacheLoader(this, key, Mono.defer(() -> getNow(key)));
+    }
+
     @Override
     @SuppressWarnings("all")
     public final Mono<E> getMono(Object key) {
-        return (Mono<E>) cacheLoading
-            .computeIfAbsent(key, _key -> new CacheLoader(this, _key, getNow(_key)))
+        return Mono.defer(() -> (Mono<E>) cacheLoading.computeIfAbsent(key, this::createLoader))
             .onErrorResume(err -> handleLoaderError(key, err));
     }
 
@@ -110,7 +157,7 @@ public final Mono<E> getMono(Object key, Supplier<Mono<E>> loader) {
         return Mono
             .deferContextual(ctx -> {
                 CacheLoader cacheLoader = cacheLoading.compute(key, (_key, old) -> {
-                    CacheLoader cl = new CacheLoader(this, _key, getNow(_key));
+                    CacheLoader cl = createLoader(_key);
                     cl.defaultValue(loader.get(), ctx);
                     return cl;
                 });
@@ -122,16 +169,16 @@ public final Mono<E> getMono(Object key, Supplier<Mono<E>> loader) {
 
     @Override
     public final Flux<E> getFlux(Object key) {
-        return (cacheLoading.computeIfAbsent(key, _key -> new CacheLoader(this, _key, getNow(_key))))
-            .flatMapIterable(e -> ((List<E>) e))
+        return Flux.defer(() -> cacheLoading.computeIfAbsent(key, this::createLoader)
+                                           .flatMapIterable(e -> ((List<E>) e)))
             .onErrorResume(err -> handleLoaderError(key, err));
     }
 
     @Override
     public final Flux<E> getFlux(Object key, Supplier<Flux<E>> loader) {
         return Flux.deferContextual(ctx -> {
                        CacheLoader cacheLoader = cacheLoading.compute(key, (_key, old) -> {
-                           CacheLoader cl = new CacheLoader(this, _key, getNow(_key));
+                           CacheLoader cl = createLoader(_key);
                            cl.defaultValue(loader.get().collectList(), ctx);
                            return cl;
                        });
@@ -141,9 +188,17 @@ public final Flux<E> getFlux(Object key, Supplier<Flux<E>> loader) {
     }
 
     protected Mono<E> handleLoaderError(Object key, Throwable err) {
-        log.warn("load cache error,key:{},evict it.", key, err);
-        return evict(key)
-            .then(Mono.empty());
+        return handleLoaderError(key, err, evict(key));
+    }
+
+    protected Mono<Void> evictLoading(Object key, CacheLoader owner) {
+        // 异步后端保留原失效逻辑；本地后端在同一原子操作中校验身份和清理。
+        return evict(key);
+    }
+
+    private <V> Mono<V> handleLoaderError(Object key, Throwable err, Mono<Void> cleanup) {
+        log.warn("load cache error,key:{}.", key, err);
+        return cleanup.then(Mono.empty());
     }
 
     @Override

diff --git a/...rent-cache/src/main/java/org/hswebframework/web/cache/supports/CaffeineReactiveCache.java b/...rent-cache/src/main/java/org/hswebframework/web/cache/supports/CaffeineReactiveCache.java
@@ -10,6 +10,9 @@
 import java.util.Arrays;
 import java.util.Collection;
 
+/**
+ * Caffeine 的响应式适配，存储操作同步执行，以在途加载者身份保护本地失效。
+ */
 @SuppressWarnings("all")
 @AllArgsConstructor
 public class CaffeineReactiveCache<E> extends AbstractReactiveCache<E> {
@@ -18,7 +21,7 @@ public class CaffeineReactiveCache<E> extends AbstractReactiveCache<E> {
 
     @Override
     public Mono<Void> evictAll(Iterable<?> key) {
-        return Mono.fromRunnable(() -> cache.invalidateAll(key));
+        return Mono.fromRunnable(() -> key.forEach(k -> invalidateLoading(k, () -> cache.invalidate(k))));
     }
 
     @Override
@@ -46,11 +49,16 @@ public Mono<Void> putNow(Object key, Object value) {
 
     @Override
     public Mono<Void> evict(Object key) {
-        return Mono.fromRunnable(() -> cache.invalidate(key));
+        return Mono.fromRunnable(() -> invalidateLoading(key, () -> cache.invalidate(key)));
+    }
+
+    @Override
+    protected Mono<Void> evictLoading(Object key, CacheLoader owner) {
+        return Mono.fromRunnable(() -> invalidateLoading(key, owner, () -> cache.invalidate(key)));
     }
 
     @Override
     public Mono<Void> clear() {
-        return Mono.fromRunnable(() -> cache.invalidateAll());
+        return Mono.fromRunnable(() -> invalidateAllLoading(cache::invalidateAll));
     }
 }
diff --git a/...current-cache/src/main/java/org/hswebframework/web/cache/supports/GuavaReactiveCache.java b/...current-cache/src/main/java/org/hswebframework/web/cache/supports/GuavaReactiveCache.java
@@ -7,6 +7,9 @@
 
 import java.util.Arrays;
 
+/**
+ * Guava 的响应式适配，存储操作同步执行，以在途加载者身份保护本地失效。
+ */
 @SuppressWarnings("all")
 @AllArgsConstructor
 public class GuavaReactiveCache<E> extends AbstractReactiveCache<E> {
@@ -16,7 +19,7 @@ public class GuavaReactiveCache<E> extends AbstractReactiveCache<E> {
 
     @Override
     public Mono<Void> evictAll(Iterable<?> key) {
-        return Mono.fromRunnable(() -> cache.invalidateAll(key));
+        return Mono.fromRunnable(() -> key.forEach(k -> invalidateLoading(k, () -> cache.invalidate(k))));
     }
 
     @Override
@@ -32,7 +35,12 @@ public Mono<Void> putNow(Object key, Object value) {
 
     @Override
     public Mono<Void> evict(Object key) {
-        return Mono.fromRunnable(() -> cache.invalidate(key));
+        return Mono.fromRunnable(() -> invalidateLoading(key, () -> cache.invalidate(key)));
+    }
+
+    @Override
+    protected Mono<Void> evictLoading(Object key, CacheLoader owner) {
+        return Mono.fromRunnable(() -> invalidateLoading(key, owner, () -> cache.invalidate(key)));
     }
 
     @Override
@@ -51,6 +59,6 @@ public Flux<E> getAll(Object... keys) {
 
     @Override
     public Mono<Void> clear() {
-        return Mono.fromRunnable(() -> cache.invalidateAll());
+        return Mono.fromRunnable(() -> invalidateAllLoading(cache::invalidateAll));
     }
 }