소스 검색

refactor(cache): CaseDataCache 案件级缓存迁移 Redis,移除 BUCKETS 与读写锁

- BUCKETS(caseId→CaseBucket)+ReentrantReadWriteLock 整体删除,改为 Redis 存储:
  正向 pno:{libType}:{人名}→SET、反查 val2name→HASH(O(1) 替代全表线性扫描)、pno-idx 清理索引
- 公开静态 API 不变,20+ 调用方零改动;putPerLib 3 命令 pipeline 为 1 次往返
- 保留原语义:开户信息重载只进正向索引;GOVERN_CONFIG 仍为本地 Map;删除无调用方的 cleanAll
- 降级策略:读失败降级空结果,写失败仅记日志不中断清洗任务
- 测试重写:Redis 用例不可达自动跳过,连 dev Redis 全部通过(7/7)
cc 5 일 전
부모
커밋
af889d73c3

+ 213 - 161
ai-server/src/main/java/com/zsjz/ai/common/cache/CaseDataCache.java

@@ -4,7 +4,6 @@ import cn.hutool.core.text.StrPool;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.extra.spring.SpringUtil;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
-import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Maps;
 import com.zsjz.ai.common.constants.StrConsts;
 import com.zsjz.ai.common.context.CaseContextHolder;
@@ -12,33 +11,52 @@ import com.zsjz.ai.common.model.person.entity.PersonLibNo;
 import com.zsjz.ai.module.dm.clean.CleanCache;
 import com.zsjz.ai.module.person.mapper.PersonLibNoMapper;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
 
+import java.nio.charset.StandardCharsets;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 /**
- * 案件数据缓存:人员库号(手机号 / 银行卡)映射。
+ * 案件数据缓存:人员库号(手机号 / 银行卡 / 证件号)映射,存 Redis,按 caseId 隔离。
  *
- * <p><b>改造说明</b>:原先是全静态的单份缓存,数据来源是<b>案件库</b>的 {@code person_lib_no} 表。
- * 服务端多人同时使用时,A 用户开案会把 B 用户的人员库缓存清掉并换成 A 的案件数据,
- * 表现为"查别人案件的人却查出了自己案件的结果"。现在改为<b>按 caseId 隔离</b>。
+ * <p><b>改造说明</b>:原先是 JVM 内 {@code BUCKETS}(caseId → CaseBucket)加每桶一把
+ * {@code ReentrantReadWriteLock} 的写法,锁域统一、快照重建等逻辑复杂。现整体迁到 Redis:
+ * Redis 单命令天然原子,不需要任何锁,代码只剩查库与读写 Redis 两件事。
  *
- * <p><b>锁域统一</b>:bucket 内统一用 {@link ReentrantReadWriteLock} 保护——
- * 原先 initCache 用 {@code synchronized(bucket)}、initPersonLibNoCache 无锁批量 put、
- * 部分读方法不加读锁,三套控制互不互斥,开案初始化期间并发读 multimap 会 CME。
- * 现在:重建在<b>锁外</b>构建完整新 multimap,写锁内<b>整体替换引用</b>;
- * 所有读走读锁并在锁内拷贝结果;不存在半成品可见窗口。
+ * <p><b>key 布局</b>(前缀复用 {@link StrConsts#CACHE_TAG_QING_JIAN},管理端
+ * {@code SystemService.cleanCache()} 的 {@code KEYS qing_jian*} 全量清理自动覆盖;
+ * {@code {caseId}} 花括号是 Redis Cluster hash tag,将来上集群时同案 key 落同一 slot):
+ * <ul>
+ *   <li>{@code qing_jian:case:{caseId}:pno:{libType}:{personName}} —— SET&lt;personNo&gt;,
+ *       正向查询:getPerPhones / getPerCards</li>
+ *   <li>{@code qing_jian:case:{caseId}:val2name} —— HASH&lt;personNo, personName&gt;,
+ *       反向查询:getPersonPhone / getPersonCard / findPersonVal / containsPersonValue,
+ *       O(1) 反查替代原先 findPersonVal 的全表线性扫描。原 personPhone/personCard 只收
+ *       libType 2/3,val2name 收全部类型;手机 / 卡 / 证件号位数互不重叠,不会互相干扰</li>
+ *   <li>{@code qing_jian:case:{caseId}:pno-idx} —— SET&lt;正向 key&gt;,清理索引:
+ *       cleanCache / initCache 据此精确删除,避免 SCAN</li>
+ * </ul>
  *
- * <p>{@link #GOVERN_CONFIG} 保持全局:它的 key 是
- * {@code StrConsts.cache_get_call_table_name()} 等(已拼接案件 ID),天然按案件隔离,
- * 清理时须用 {@link #cleanGovernConfig(Long)} 按 key 精确匹配,禁止全表 clear。
+ * <p><b>降级策略</b>:读方法 Redis 异常时记 error 日志并降级返回空结果(不中断业务);
+ * putPerLib 写失败也只记日志——清洗任务一跑几十分钟,不能因缓存抖动被杀掉,
+ * 数据仍会经 CleanCache 缓冲落库,下次 initCache 重建时修复缓存;
+ * initCache 失败向上抛(CaseInfoService.open 已有兜底捕获)。
+ * 重建窗口内的并发 putPerLib 可能丢写,与原实现"快照换引用"的竞态同级,不是回归。
  *
- * <p>案件 ID 从 {@link CaseContextHolder#get()} 解析。读方法取不到时降级返回空集并打 error 日志
- * (不中断业务),写方法取不到时直接抛异常(避免把数据写进别人的缓存)。
+ * <p>数据不设 TTL:正常路径由退案 / 换案 / 删案 / 会话结束的 {@link #cleanCache(Long)}
+ * 清理;进程异常退出最多遗留"案件数"组孤儿 key,由管理端全量清理兜底。
+ *
+ * <p>{@link #GOVERN_CONFIG} 保持本地:key 已含案件 ID(前缀直拼),且只有
+ * 表名级别的两三条,本地 ConcurrentHashMap 足够,清理时须用
+ * {@link #cleanGovernConfig(Long)} 按 key 精确匹配,禁止全表 clear。
+ *
+ * <p>案件 ID 从 {@link CaseContextHolder#get()} 解析。写方法取不到时直接抛异常
+ * (避免把数据写进别的案件),读方法取不到时降级返回空集合并打 error 日志。
  */
 @Slf4j
 public class CaseDataCache {
@@ -49,183 +67,138 @@ public class CaseDataCache {
      */
     public static final Map<String, String> GOVERN_CONFIG = Maps.newConcurrentMap();
 
+    private static final String CASE_PREFIX = StrConsts.CACHE_TAG_QING_JIAN + ":case:";
+    private static final String FORWARD = ":pno:";
+    private static final String VAL2NAME = ":val2name";
+    private static final String IDX = ":pno-idx";
+
     /**
-     * caseId → 案件级缓存
+     * initCache 重建时每批 pipeline 写入的行数:单批命令数 = 行数 × 3,控制在千级避免
+     * Lettuce 客户端缓冲过大。
      */
-    private static final Map<Long, CaseBucket> BUCKETS = new ConcurrentHashMap<>();
-
-    private CaseDataCache() {
-    }
+    private static final int INIT_BATCH = 1000;
 
     /**
-     * 单个案件的人员库缓存。
-     * <p>三个容器字段均<b>非 final</b>:initCache 在写锁内整体替换引用(快照式重建)。
+     * 懒取的 Redis 模板。并发重复取同一个单例 bean 无害,不值得为此加锁。
      */
-    private static final class CaseBucket {
-        private HashMultimap<String, String> personLibNoMap = HashMultimap.create();
-        private Map<String, String> personCard = Maps.newConcurrentMap();
-        private Map<String, String> personPhone = Maps.newConcurrentMap();
-        private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
-        private final ReentrantReadWriteLock.ReadLock readLock = rwLock.readLock();
-        private final ReentrantReadWriteLock.WriteLock writeLock = rwLock.writeLock();
+    private static volatile StringRedisTemplate REDIS;
+
+    private CaseDataCache() {
     }
 
     /**
-     * 初始化指定案件的人员库缓存(开案时调用)。
+     * 初始化指定案件的人员库缓存(开案 / 手工维护人员库后调用)。
      *
-     * <p>查库与构建都在锁外完成,写锁内只做整体替换引用——
-     * 重建期间并发的读写要么看到旧快照、要么看到新快照,永远不会看到半成品。
+     * <p>先按 pno-idx 精确删旧 key,再分批 pipeline 写入。失败向上抛;
+     * 重建窗口内读到的可能是空 / 半份数据——它只是缓存,下次重建即修复。
      */
     public static void initCache(Long caseId) {
         Long resolved = requireCaseId(caseId, "initCache");
-        CaseBucket bucket = BUCKETS.computeIfAbsent(resolved, key -> new CaseBucket());
         // 显式绑定案件上下文:查的是案件库,@DS("slave") 需要知道是哪个案件。
-        // DB 调用不进锁
         PersonLibNoMapper mapper = SpringUtil.getBean(PersonLibNoMapper.class);
         List<PersonLibNo> personLibNos = CaseContextHolder.callWith(resolved,
                 () -> mapper.selectList(Wrappers.emptyWrapper()));
-        HashMultimap<String, String> newMultimap = HashMultimap.create();
-        Map<String, String> newPhone = Maps.newConcurrentMap();
-        Map<String, String> newCard = Maps.newConcurrentMap();
-        for (PersonLibNo personLibNo : personLibNos) {
-            String key = personLibNo.getPersonName() + StrPool.UNDERLINE + personLibNo.getLibType();
-            newMultimap.put(key, personLibNo.getPersonNo());
-            Integer libType = personLibNo.getLibType();
-            if (Integer.valueOf(2).equals(libType)) {
-                newPhone.put(personLibNo.getPersonNo(), personLibNo.getPersonName());
-            } else if (Integer.valueOf(3).equals(libType)) {
-                newCard.put(personLibNo.getPersonNo(), personLibNo.getPersonName());
-            }
-        }
-        bucket.writeLock.lock();
-        try {
-            bucket.personLibNoMap = newMultimap;
-            bucket.personPhone = newPhone;
-            bucket.personCard = newCard;
-        } finally {
-            bucket.writeLock.unlock();
+        deleteCaseKeys(resolved);
+        byte[] val2name = utf8(val2nameKey(resolved));
+        byte[] idx = utf8(idxKey(resolved));
+        for (int from = 0; from < personLibNos.size(); from += INIT_BATCH) {
+            List<PersonLibNo> batch = personLibNos.subList(from,
+                    Math.min(from + INIT_BATCH, personLibNos.size()));
+            redis().executePipelined((RedisCallback<Object>) conn -> {
+                for (PersonLibNo personLibNo : batch) {
+                    // 反查 hash 不接受 null field/value,脏数据直接跳过
+                    if (personLibNo.getPersonName() == null || personLibNo.getPersonNo() == null) {
+                        continue;
+                    }
+                    byte[] forward = utf8(forwardKey(resolved, personLibNo.getLibType(),
+                            personLibNo.getPersonName()));
+                    byte[] no = utf8(personLibNo.getPersonNo());
+                    conn.setCommands().sAdd(forward, no);
+                    conn.hashCommands().hSet(val2name, no, utf8(personLibNo.getPersonName()));
+                    conn.setCommands().sAdd(idx, forward);
+                }
+                return null;
+            });
         }
     }
 
     /**
-     * 新增人员库
+     * 新增人员库(通话 / 流水清洗逐条调用,热路径)。
+     * 3 条命令 pipelined 为 1 次网络往返。Redis 写失败只记日志不抛:
+     * 数据仍会经 CleanCache 缓冲落库,下次 initCache 重建时修复缓存。
      */
     public static void putPerLib(String key, String value, String bankName) {
-        CaseBucket bucket = requireBucket("putPerLib");
-        bucket.writeLock.lock();
-        try {
-            bucket.personLibNoMap.put(key, value);
-            String[] keys = StrUtil.splitToArray(key, StrPool.UNDERLINE);
-            int i = Integer.parseInt(keys[1]);
-            if (i == 2) {
-                bucket.personPhone.put(value, keys[0]);
-            } else if (i == 3) {
-                bucket.personCard.put(value, keys[0]);
-            }
-            CleanCache.addPersonLibNo(new PersonLibNo(keys[0], value, Integer.parseInt(keys[1]), 1, bankName));
-        } finally {
-            // 必须在 finally 块中释放锁,确保即使发生异常也能正确释放
-            bucket.writeLock.unlock();
-        }
+        Long caseId = requireCaseId(null, "putPerLib");
+        String[] keys = StrUtil.splitToArray(key, StrPool.UNDERLINE);
+        String personName = keys[0];
+        int libType = Integer.parseInt(keys[1]);
+        putEntry(caseId, personName, libType, value, true);
+        CleanCache.addPersonLibNo(new PersonLibNo(personName, value, libType, 1, bankName));
     }
 
+    /**
+     * 新增人员库(开户信息清洗)。
+     * <p>与原实现一致:只进正向索引,不写 val2name 反查——
+     * 开户信息里的姓名可能不规范,避免污染"号码 → 人名"反查。
+     */
     public static void putPerLib(String key, String value, LocalDateTime startTime, LocalDateTime endTime, String operatorId) {
-        CaseBucket bucket = requireBucket("putPerLib");
-        bucket.writeLock.lock();
-        try {
-            bucket.personLibNoMap.put(key, value);
-            String[] keys = StrUtil.splitToArray(key, StrPool.UNDERLINE);
-            CleanCache.addPersonLibNo(new PersonLibNo(keys[0], value, Integer.parseInt(keys[1]), startTime, endTime, operatorId));
-        } finally {
-            bucket.writeLock.unlock();
-        }
+        Long caseId = requireCaseId(null, "putPerLib");
+        String[] keys = StrUtil.splitToArray(key, StrPool.UNDERLINE);
+        String personName = keys[0];
+        int libType = Integer.parseInt(keys[1]);
+        putEntry(caseId, personName, libType, value, false);
+        CleanCache.addPersonLibNo(new PersonLibNo(personName, value, libType, startTime, endTime, operatorId));
     }
 
     /**
-     * 获取人的所有手机号(锁内拷贝,返回不可变集合)
+     * 获取人的所有手机号
      */
     public static Set<String> getPerPhones(String personName) {
-        CaseBucket bucket = bucketOrNull("getPerPhones");
-        if (bucket == null) {
-            return Set.of();
-        }
-        bucket.readLock.lock();
-        try {
-            return Set.copyOf(bucket.personLibNoMap.get(personName + StrPool.UNDERLINE + 2));
-        } finally {
-            bucket.readLock.unlock();
-        }
+        return perNos(personName, 2, "getPerPhones");
     }
 
     /**
-     * 获取人的所有银行卡(锁内拷贝,返回不可变集合)
+     * 获取人的所有银行卡
      */
     public static Set<String> getPerCards(String personName) {
-        CaseBucket bucket = bucketOrNull("getPerCards");
-        if (bucket == null) {
-            return Set.of();
-        }
-        bucket.readLock.lock();
-        try {
-            return Set.copyOf(bucket.personLibNoMap.get(personName + StrPool.UNDERLINE + 3));
-        } finally {
-            bucket.readLock.unlock();
-        }
+        return perNos(personName, 3, "getPerCards");
     }
 
     /**
      * 按手机号取人名
      */
     public static String getPersonPhone(String phone) {
-        CaseBucket bucket = bucketOrNull("getPersonPhone");
-        return bucket == null ? null : bucket.personPhone.get(phone);
+        return nameOf(phone, "getPersonPhone");
     }
 
     /**
      * 按银行卡号取人名
      */
     public static String getPersonCard(String card) {
-        CaseBucket bucket = bucketOrNull("getPersonCard");
-        return bucket == null ? null : bucket.personCard.get(card);
+        return nameOf(card, "getPersonCard");
     }
 
     /**
-     * 该值(手机号/卡号)是否属于本案件人员库
+     * 该值(手机号/卡号/证件号)是否属于本案件人员库
      */
     public static boolean containsPersonValue(String val) {
-        CaseBucket bucket = bucketOrNull("containsPersonValue");
-        if (bucket == null) {
+        Long caseId = caseIdOrNull("containsPersonValue");
+        if (caseId == null) {
             return false;
         }
-        bucket.readLock.lock();
         try {
-            return bucket.personLibNoMap.containsValue(val);
-        } finally {
-            bucket.readLock.unlock();
+            return Boolean.TRUE.equals(redis().opsForHash().hasKey(val2nameKey(caseId), val));
+        } catch (Exception e) {
+            log.error("CaseDataCache.containsPersonValue 读 Redis 失败,本次降级返回 false: caseId={}", caseId, e);
+            return false;
         }
     }
 
+    /**
+     * 按号码反查人名(O(1),替代原先全表线性扫描)
+     */
     public static String findPersonVal(String val) {
-        CaseBucket bucket = bucketOrNull("findPersonVal");
-        if (bucket == null) {
-            return null;
-        }
-        bucket.readLock.lock();
-        try {
-            if (bucket.personLibNoMap.containsValue(val)) {
-                for (Map.Entry<String, String> entry : bucket.personLibNoMap.entries()) {
-                    String key = entry.getKey();
-                    String value = entry.getValue();
-                    if (val.equals(value)) {
-                        String[] keys = key.split(StrPool.UNDERLINE);
-                        return keys[0];
-                    }
-                }
-            }
-            return null;
-        } finally {
-            bucket.readLock.unlock();
-        }
+        return nameOf(val, "findPersonVal");
     }
 
     /**
@@ -250,28 +223,107 @@ public class CaseDataCache {
     }
 
     /**
-     * 清理指定案件的缓存(退出案件 / 换案时调用)
+     * 清理指定案件的缓存(退出案件 / 换案 / 删案 / 会话结束时调用)。
+     * Redis 异常只记日志不抛:调用点都在 listener / reaper 里,清理失败不该打断它们。
      */
     public static void cleanCache(Long caseId) {
         if (caseId == null) {
             return;
         }
-        BUCKETS.remove(caseId);
+        try {
+            deleteCaseKeys(caseId);
+        } catch (Exception e) {
+            log.error("CaseDataCache.cleanCache 清理 Redis 失败: caseId={}", caseId, e);
+        }
+    }
+
+    // ------------------------------------------------------------------
+    // 内部工具
+    // ------------------------------------------------------------------
+
+    private static StringRedisTemplate redis() {
+        StringRedisTemplate template = REDIS;
+        if (template == null) {
+            template = REDIS = SpringUtil.getBean(StringRedisTemplate.class);
+        }
+        return template;
+    }
+
+    /** 测试注入用(同包测试类绕过 Spring 容器直连 Redis),生产代码不要调用 */
+    static void useRedisTemplate(StringRedisTemplate template) {
+        REDIS = template;
     }
 
     /**
-     * 清理全部案件的缓存(应用停止时调用)
+     * 写一条人员库映射:SADD 正向 + (可选)HSET 反查 + SADD 清理索引,
+     * pipelined 成 1 次往返。
      */
-    public static void cleanAll() {
-        BUCKETS.clear();
+    private static void putEntry(Long caseId, String personName, int libType, String personNo,
+                                 boolean withReverse) {
+        byte[] forward = utf8(forwardKey(caseId, libType, personName));
+        byte[] val2name = utf8(val2nameKey(caseId));
+        byte[] idx = utf8(idxKey(caseId));
+        byte[] no = utf8(personNo);
+        byte[] name = utf8(personName);
+        try {
+            redis().executePipelined((RedisCallback<Object>) conn -> {
+                conn.setCommands().sAdd(forward, no);
+                if (withReverse) {
+                    conn.hashCommands().hSet(val2name, no, name);
+                }
+                conn.setCommands().sAdd(idx, forward);
+                return null;
+            });
+        } catch (Exception e) {
+            log.error("CaseDataCache.putPerLib 写 Redis 失败(不中断业务,落库缓冲不受影响): caseId={}, personName={}, libType={}",
+                    caseId, personName, libType, e);
+        }
     }
 
-    // ------------------------------------------------------------------
-    // 内部工具
-    // ------------------------------------------------------------------
+    /** 正向查询:人名 + libType → 号码集合 */
+    private static Set<String> perNos(String personName, int libType, String action) {
+        Long caseId = caseIdOrNull(action);
+        if (caseId == null) {
+            return Set.of();
+        }
+        try {
+            Set<String> values = redis().opsForSet().members(forwardKey(caseId, libType, personName));
+            return values == null ? Set.of() : Set.copyOf(values);
+        } catch (Exception e) {
+            log.error("CaseDataCache.{} 读 Redis 失败,本次降级返回空结果: caseId={}", action, caseId, e);
+            return Set.of();
+        }
+    }
+
+    /** 反向查询:号码 → 人名 */
+    private static String nameOf(String personNo, String action) {
+        Long caseId = caseIdOrNull(action);
+        if (caseId == null) {
+            return null;
+        }
+        try {
+            Object name = redis().opsForHash().get(val2nameKey(caseId), personNo);
+            return name == null ? null : name.toString();
+        } catch (Exception e) {
+            log.error("CaseDataCache.{} 读 Redis 失败,本次降级返回 null: caseId={}", action, caseId, e);
+            return null;
+        }
+    }
+
+    /** 按 pno-idx 精确删除该案件全部人员库 key(不用 SCAN) */
+    private static void deleteCaseKeys(Long caseId) {
+        Set<String> forwards = redis().opsForSet().members(idxKey(caseId));
+        List<String> keys = new ArrayList<>();
+        if (forwards != null) {
+            keys.addAll(forwards);
+        }
+        keys.add(idxKey(caseId));
+        keys.add(val2nameKey(caseId));
+        redis().delete(keys);
+    }
 
     private static Long requireCaseId(Long explicit, String action) {
-        Long caseId = resolveCaseId(explicit);
+        Long caseId = explicit != null ? explicit : CaseContextHolder.get();
         if (caseId == null) {
             throw new IllegalStateException(
                     "CaseDataCache." + action + " 缺少案件上下文:异步线程必须用 CaseContextHolder.runWith/callWith 显式绑定案件 ID");
@@ -279,30 +331,30 @@ public class CaseDataCache {
         return caseId;
     }
 
-    /**
-     * 写操作用:取不到案件上下文直接抛异常
-     */
-    private static CaseBucket requireBucket(String action) {
-        Long caseId = requireCaseId(null, action);
-        return BUCKETS.computeIfAbsent(caseId, key -> new CaseBucket());
-    }
-
     /**
      * 读操作用:取不到案件上下文时打日志并返回 null,由调用方降级
      */
-    private static CaseBucket bucketOrNull(String action) {
-        Long caseId = resolveCaseId(null);
+    private static Long caseIdOrNull(String action) {
+        Long caseId = CaseContextHolder.get();
         if (caseId == null) {
             log.error("CaseDataCache.{} 缺少案件上下文,本次读取降级返回空结果", action);
-            return null;
         }
-        return BUCKETS.get(caseId);
+        return caseId;
     }
 
-    private static Long resolveCaseId(Long explicit) {
-        if (explicit != null) {
-            return explicit;
-        }
-        return CaseContextHolder.get();
+    private static String forwardKey(Long caseId, Integer libType, String personName) {
+        return CASE_PREFIX + caseId + FORWARD + libType + ':' + personName;
+    }
+
+    private static String val2nameKey(Long caseId) {
+        return CASE_PREFIX + caseId + VAL2NAME;
+    }
+
+    private static String idxKey(Long caseId) {
+        return CASE_PREFIX + caseId + IDX;
+    }
+
+    private static byte[] utf8(String s) {
+        return s.getBytes(StandardCharsets.UTF_8);
     }
 }

+ 157 - 114
ai-server/src/test/java/com/zsjz/ai/common/cache/CaseDataCacheConcurrencyTest.java

@@ -1,9 +1,16 @@
 package com.zsjz.ai.common.cache;
 
 import com.zsjz.ai.common.context.CaseContextHolder;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
 
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -11,8 +18,8 @@ import java.util.Set;
 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.AtomicReference;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -21,23 +28,65 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
- * CaseDataCache 并发回归测试(P0-7 + P1-17)。
+ * CaseDataCache 回归测试(Redis 版)。
  *
- * <p>不依赖 Spring/DB:initCache 需要 mapper(SpringUtil 取 Bean),离线单测不覆盖;
- * 这里覆盖纯内存路径——putPerLib 写锁与各读锁的并发正确性、按案件隔离、
- * cleanGovernConfig 按 key 精确清除(回归「startsWith 前缀误伤 caseId=123」的陷阱)。
+ * <p>覆盖三部分:
+ * <ol>
+ *   <li>GOVERN_CONFIG(本地 Map)cleanGovernConfig 按 key 精确清除——离线可跑;</li>
+ *   <li>缺案件上下文:写抛异常、读降级——离线可跑(不触 Redis);</li>
+ *   <li>Redis 读写回环 / 案件隔离 / cleanCache 删除 / 并发写——需要真实 Redis,
+ *       不可达时自动跳过。默认连 127.0.0.1:6379,用
+ *       {@code -Dcase-cache.redis.host=... -Dcase-cache.redis.port=...} 指定
+ *       (dev 环境为 192.168.0.106:6379)。</li>
+ * </ol>
+ * initCache 需要 mapper(SpringUtil 取 Bean),离线单测不覆盖,走联调验证。
  */
 class CaseDataCacheConcurrencyTest {
 
-    /** 人员名 → 库号(putPerLib 的 key 约定:personName_libType) */
-    private static String phoneOf(long caseId, int writer, int i) {
-        return "13" + caseId + String.format("%04d", writer) + String.format("%04d", i);
+    /** 测试用案件 ID(@AfterAll 统一清理,避免污染共享 Redis) */
+    private static final long[] REDIS_CASE_IDS = {910_100L, 910_101L, 910_200L, 911_000L};
+
+    private static LettuceConnectionFactory connectionFactory;
+    private static StringRedisTemplate template;
+    private static boolean redisReady;
+
+    @BeforeAll
+    static void connectRedis() {
+        String host = System.getProperty("case-cache.redis.host", "127.0.0.1");
+        int port = Integer.parseInt(System.getProperty("case-cache.redis.port", "6379"));
+        try {
+            connectionFactory = new LettuceConnectionFactory(host, port);
+            connectionFactory.afterPropertiesSet();
+            template = new StringRedisTemplate(connectionFactory);
+            template.afterPropertiesSet();
+            try (RedisConnection conn = connectionFactory.getConnection()) {
+                conn.ping();
+            }
+            CaseDataCache.useRedisTemplate(template);
+            redisReady = true;
+        } catch (Exception e) {
+            redisReady = false;
+        }
     }
 
-    private static String nameOf(int writer, int i) {
-        return "人员w" + writer + "i" + i;
+    @AfterAll
+    static void cleanUp() {
+        for (long caseId : REDIS_CASE_IDS) {
+            try {
+                CaseDataCache.cleanCache(caseId);
+            } catch (Exception ignored) {
+                // Redis 不可达时 cleanCache 内部已降级,这里兜底不让清理失败炸掉收尾
+            }
+        }
+        if (connectionFactory != null) {
+            connectionFactory.destroy();
+        }
     }
 
+    // ------------------------------------------------------------------
+    // 离线用例:GOVERN_CONFIG(本地 Map)
+    // ------------------------------------------------------------------
+
     @Test
     @DisplayName("cleanGovernConfig 精确匹配:只清本案件条目(12 不误伤 123/124)")
     void cleanGovernConfigExactMatch() {
@@ -66,18 +115,91 @@ class CaseDataCacheConcurrencyTest {
         CaseDataCache.cleanGovernConfig(null);
     }
 
+    // ------------------------------------------------------------------
+    // 离线用例:案件上下文约束
+    // ------------------------------------------------------------------
+
+    @Test
+    @DisplayName("缺案件上下文:putPerLib 抛异常(防写进别人案件),读降级返回空")
+    void contextRequired() {
+        assertThrows(IllegalStateException.class, () ->
+                CaseDataCache.putPerLib("a_2", "v", "bank"));
+        // 未绑定上下文的读:直接降级,不触碰 Redis
+        assertEquals(Set.of(), CaseDataCache.getPerPhones("x"));
+        assertNull(CaseDataCache.getPersonPhone("x"));
+        assertFalse(CaseDataCache.containsPersonValue("x"));
+        assertNull(CaseDataCache.findPersonVal("x"));
+    }
+
+    // ------------------------------------------------------------------
+    // Redis 用例(不可达自动跳过)
+    // ------------------------------------------------------------------
+
+    @Test
+    @DisplayName("putPerLib 写读回环 + 案件隔离(Redis)")
+    void redisRoundTripAndIsolation() {
+        Assumptions.assumeTrue(redisReady, "Redis 不可达,跳过(-Dcase-cache.redis.host=... 指定地址)");
+        long caseA = REDIS_CASE_IDS[0];
+        long caseB = REDIS_CASE_IDS[1];
+
+        CaseContextHolder.runWith(caseA, 1L, () ->
+                CaseDataCache.putPerLib("张三_2", "13800000001", "移动"));
+        CaseContextHolder.runWith(caseA, 1L, () -> {
+            assertEquals("张三", CaseDataCache.getPersonPhone("13800000001"), "反查:号码→人名");
+            assertTrue(CaseDataCache.getPerPhones("张三").contains("13800000001"), "正向:人名→手机号");
+            assertTrue(CaseDataCache.containsPersonValue("13800000001"));
+            assertEquals("张三", CaseDataCache.findPersonVal("13800000001"));
+        });
+        // 跨案件互查:案件 B 查不到案件 A 写入的号码(按 caseId 隔离的回归哨兵)
+        CaseContextHolder.runWith(caseB, 1L, () -> {
+            assertNull(CaseDataCache.getPersonPhone("13800000001"), "案件 B 不应查到案件 A 的人员");
+            assertFalse(CaseDataCache.containsPersonValue("13800000001"));
+            assertTrue(CaseDataCache.getPerPhones("张三").isEmpty());
+        });
+    }
+
     @Test
-    @DisplayName("多案件并发 putPerLib:按案件隔离,互不串数据")
-    void multiCasePutIsolation() throws Exception {
-        int cases = 4;
-        int writersPerCase = 4;
-        int itemsPerWriter = 300;
-        ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
-        CountDownLatch start = new CountDownLatch(1);
-        List<java.util.concurrent.Future<?>> futures = new ArrayList<>();
-        for (int c = 0; c < cases; c++) {
-            long caseId = 910_000L + c;
-            for (int w = 0; w < writersPerCase; w++) {
+    @DisplayName("开户信息重载只进正向索引,不写号码→人名反查(保持原实现语义)")
+    void openInfoOverloadForwardOnly() {
+        Assumptions.assumeTrue(redisReady, "Redis 不可达,跳过");
+        long caseId = REDIS_CASE_IDS[0];
+        CaseContextHolder.runWith(caseId, 1L, () ->
+                CaseDataCache.putPerLib("王五_2", "13900000002",
+                        LocalDateTime.now(), LocalDateTime.now(), "联通"));
+        CaseContextHolder.runWith(caseId, 1L, () -> {
+            assertTrue(CaseDataCache.getPerPhones("王五").contains("13900000002"), "正向索引应有记录");
+            assertNull(CaseDataCache.getPersonPhone("13900000002"), "开户信息不进 val2name 反查");
+        });
+    }
+
+    @Test
+    @DisplayName("cleanCache:删除该案件全部 Redis key")
+    void cleanCacheRemovesCaseKeys() {
+        Assumptions.assumeTrue(redisReady, "Redis 不可达,跳过");
+        long caseId = REDIS_CASE_IDS[2];
+        CaseContextHolder.runWith(caseId, 1L, () ->
+                CaseDataCache.putPerLib("李四_3", "6222000011112222", "工商银行"));
+        CaseContextHolder.runWith(caseId, 1L, () ->
+                assertTrue(CaseDataCache.containsPersonValue("6222000011112222")));
+        CaseDataCache.cleanCache(caseId);
+        CaseContextHolder.runWith(caseId, 1L, () ->
+                assertFalse(CaseDataCache.containsPersonValue("6222000011112222"), "清理后不得再可见"));
+        assertFalse(Boolean.TRUE.equals(template.hasKey("qing_jian:case:" + caseId + ":val2name")));
+        assertFalse(Boolean.TRUE.equals(template.hasKey("qing_jian:case:" + caseId + ":pno-idx")));
+    }
+
+    @Test
+    @DisplayName("多线程并发 putPerLib:最终全部可见(Redis 命令原子性)")
+    void concurrentWritersAllVisible() throws Exception {
+        Assumptions.assumeTrue(redisReady, "Redis 不可达,跳过");
+        long caseId = REDIS_CASE_IDS[3];
+        int writers = 4;
+        int items = 100;
+        ExecutorService pool = Executors.newFixedThreadPool(writers);
+        try {
+            CountDownLatch start = new CountDownLatch(1);
+            List<Future<?>> futures = new ArrayList<>();
+            for (int w = 0; w < writers; w++) {
                 final int writer = w;
                 futures.add(pool.submit(() -> CaseContextHolder.runWith(caseId, 1L, () -> {
                     try {
@@ -86,105 +208,26 @@ class CaseDataCacheConcurrencyTest {
                         Thread.currentThread().interrupt();
                         throw new RuntimeException(e);
                     }
-                    for (int i = 0; i < itemsPerWriter; i++) {
-                        CaseDataCache.putPerLib(nameOf(writer, i) + "_2", phoneOf(caseId, writer, i), "bank");
+                    for (int i = 0; i < items; i++) {
+                        CaseDataCache.putPerLib("人员w" + writer + "_2",
+                                "13" + caseId + String.format("%02d%04d", writer, i), null);
                     }
                 })));
             }
+            start.countDown();
+            for (Future<?> f : futures) {
+                f.get(60, TimeUnit.SECONDS);
+            }
+        } finally {
+            pool.shutdown();
         }
-        start.countDown();
-        for (java.util.concurrent.Future<?> f : futures) {
-            f.get(30, TimeUnit.SECONDS);
-        }
-        for (int c = 0; c < cases; c++) {
-            long caseId = 910_000L + c;
-            CaseContextHolder.runWith(caseId, 1L, () -> {
-                for (int w = 0; w < writersPerCase; w++) {
-                    String sample = phoneOf(caseId, w, 0);
-                    assertEquals(nameOf(w, 0), CaseDataCache.findPersonVal(sample), "本案件写入的号码应能反查");
-                    assertTrue(CaseDataCache.containsPersonValue(sample));
-                    assertTrue(CaseDataCache.getPerPhones(nameOf(w, 0)).contains(sample));
-                    assertEquals(nameOf(w, 0), CaseDataCache.getPersonPhone(sample));
-                }
-            });
-        }
-        // 跨案件互查:案件 A 查不到案件 B 写入的号码(按 caseId 隔离的回归哨兵)
-        CaseContextHolder.runWith(910_001L, 1L, () -> {
-            String otherCasePhone = phoneOf(910_000L, 0, 0);
-            assertNull(CaseDataCache.findPersonVal(otherCasePhone), "案件 910001 不应查到案件 910000 的人员");
-            assertFalse(CaseDataCache.containsPersonValue(otherCasePhone));
-        });
-        // 清理测试案件的 bucket,避免污染同 JVM 其他测试
-        for (int c = 0; c < cases; c++) {
-            CaseDataCache.cleanCache(910_000L + c);
-        }
-        pool.shutdown();
-    }
-
-    @Test
-    @DisplayName("并发读(读锁)与写(写锁)竞争:无 CME,最终读全部可见")
-    void concurrentReadersAndWriters() throws Exception {
-        long caseId = 911_000L;
-        AtomicReference<Throwable> failure = new AtomicReference<>();
-        ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
-        CountDownLatch start = new CountDownLatch(1);
-        int writers = 2;
-        int readers = 4;
-        int items = 2000;
-        List<java.util.concurrent.Future<?>> futures = new ArrayList<>();
-        for (int w = 0; w < writers; w++) {
-            futures.add(pool.submit(() -> CaseContextHolder.runWith(caseId, 1L, () -> {
-                try {
-                    start.await();
-                    for (int i = 0; i < items; i++) {
-                        CaseDataCache.putPerLib(nameOf(0, i) + "_2", "137" + String.format("%011d", i), "bank");
-                    }
-                } catch (Throwable t) {
-                    failure.set(t);
-                }
-            })));
-        }
-        for (int r = 0; r < readers; r++) {
-            futures.add(pool.submit(() -> CaseContextHolder.runWith(caseId, 1L, () -> {
-                try {
-                    start.await();
-                    for (int i = 0; i < 5000; i++) {
-                        // 读锁路径:getPerPhones / containsPersonValue / findPersonVal 全走 readLock
-                        CaseDataCache.getPerPhones(nameOf(0, i % items));
-                        CaseDataCache.containsPersonValue("137" + String.format("%011d", i % items));
-                        CaseDataCache.findPersonVal("137" + String.format("%011d", i % items));
-                    }
-                } catch (Throwable t) {
-                    failure.set(t);
-                }
-            })));
-        }
-        start.countDown();
-        for (java.util.concurrent.Future<?> f : futures) {
-            f.get(60, TimeUnit.SECONDS);
-        }
-        assertNull(failure.get(), "读写并发不应抛任何异常(CME 回归哨兵)");
-        // 最终一致:所有写入都可见
         CaseContextHolder.runWith(caseId, 1L, () -> {
-            for (int i = 0; i < items; i++) {
-                assertTrue(CaseDataCache.containsPersonValue("137" + String.format("%011d", i)),
-                        "写完后每条号码都应可见,丢失说明写锁/读锁域不一致");
+            for (int w = 0; w < writers; w++) {
+                for (int i = 0; i < items; i++) {
+                    assertTrue(CaseDataCache.containsPersonValue("13" + caseId + String.format("%02d%04d", w, i)),
+                            "并发写入的每条号码都应可见,丢失说明 pipeline 写入有缺陷");
+                }
             }
         });
-        CaseDataCache.cleanCache(caseId);
-        pool.shutdown();
-    }
-
-    @Test
-    @DisplayName("缺案件上下文:putPerLib 抛异常(防写进别人案件),读降级返回空")
-    void contextRequired() {
-        assertThrows(IllegalStateException.class, () ->
-                CaseDataCache.putPerLib("a_2", "v", "bank"));
-        CaseContextHolder.runWith(912_000L, 1L, () -> {
-            assertEquals(Set.of(), CaseDataCache.getPerPhones("x"));
-            assertNull(CaseDataCache.getPersonPhone("x"));
-            assertFalse(CaseDataCache.containsPersonValue("x"));
-            assertNull(CaseDataCache.findPersonVal("x"));
-        });
     }
 }