easy-store 配置存储组件设计

本文档描述 easy-store 组件的设计与实现。该组件提供抽象化的配置存储抽象层，底层可插拔多种持久化后端（文件 / Redis / 内存），支持 CRUD + 缓存 + 配置变更监听。设计采用三层架构：接口层（ConfigStorage）→ 抽象基类层（AbstractConfigStorage / AbstractWatchableConfigStorage）→ 具体实现层（DefaultConfigStorage），以及独立的持久化后端 SPI（ConfigRepository）。

• 一、背景与问题
• 二、组件定位与边界
• 三、总体架构
• 四、核心 API 草图
• 五、包结构与核心类
• 六、缓存与监听机制
• 七、关键设计决策
• 八、已知问题与待改进

一、背景与问题

1.1 为什么需要统一存储抽象

Matrix Framework 中多个引擎组件需要配置持久化：

如果每个引擎都自己实现文件读/Redis 读，会导致：

• 重复代码（每个引擎都要实现 load/save/delete + 缓存 + 命名空间）
• 后端切换困难（从文件切到 Redis 要改引擎内部代码）
• 缺少统一的配置变更监听机制

easy-store 把存储抽象成 ConfigStorage<T> 接口 + ConfigRepository<T> SPI，引擎注入 ConfigStorage 即可，后端通过 Repository 实现灵活切换。

1.2 设计来源

easy-store 是旧 easy-flows 拆分出的独立模块。旧 easy-flows 包含了工作流引擎 + 规则引擎 + 存储抽象三部分，新 easy-flows 聚焦工作流引擎（基于 warm-flow），存储能力独立为 easy-store。

二、组件定位与边界

定位：可插拔后端的配置存储抽象层，提供统一 CRUD + 缓存 + 监听接口。

边界：

约束：

• 写路径不刷缓存：save 直接写 Repository 不更新 Cache，delete 才先清缓存——save 后立即 load 仍可能命中旧缓存（除非之前没有缓存该 key）
• 监听机制是被动的：fireConfigChangeEvent 是 protected 方法，DefaultConfigStorage 没有重写 save/delete 触发；实际使用需子类自行触发或 AOP 拦截
• RedisRepository 未完成：list() 返回空、delete(namespace) 不生效，半成品

三、总体架构

flowchart TD
    subgraph "接口层"
        A1["ConfigStorage<T extends ConfigDefinition><br/>default 方法组合 Repository + Cache + Namespace"]
        A2["WatchableConfigStorage<T><br/>extends ConfigStorage<br/>增加 watch/watchNamespace/watchKeys/unwatch"]
    end

    subgraph "抽象基类层（数据持有者）"
        B1["AbstractConfigStorage<T><br/>持有 ConfigRepository + ConcurrentMap cache + namespace"]
        B2["AbstractWatchableConfigStorage<T><br/>extends AbstractConfigStorage<br/>维护 watchers Map<br/>fireConfigChangeEvent 保护方法"]
    end

    subgraph "具体实现层"
        C1["DefaultConfigStorage<T><br/>extends AbstractWatchableConfigStorage<br/>提供 DefaultWatchHandle"]
    end

    subgraph "持久化后端 SPI"
        D1["ConfigRepository<T><br/>接口：find/list/save/delete(namespace)/delete(namespace,key)"]
        D2["InMemoryRepository<br/>后端：ConcurrentHashMap"]
        D3["FileRepository<br/>后端：YAML 文件（classpath 或可写目录）<br/>⚠️ 非泛型，硬编码 ConfigDefinition"]
        D4["RedisRepository<br/>后端：RedisTemplate<br/>⚠️ list/delete(namespace) 未实现"]
    end

    subgraph "模型层"
        E1["ConfigDefinition<br/>POJO：namespace + key + text + version + attachment"]
        E2["ConfigChangeEvent<T><br/>变更事件：oldValue + newValue + ChangeType(CREATE/UPDATE/DELETE)"]
        E3["ConfigChangeListener<T><br/>@FunctionalInterface：onChange(ConfigChangeEvent)"]
        E4["WatchHandle<br/>接口：watchId + isActive + cancel"]
    end

    A1 --> B1
    A2 --> B2
    B1 --> B2
    B2 --> C1
    B1 --> D1
    D1 --> D2
    D1 --> D3
    D1 --> D4

ConfigStorage default 方法的协作流：

flowchart TD
    subgraph "读路径"
        R1["load(namespace, key)"]
        R2["cache.computeIfAbsent()"]
        R3{"缓存命中?"}
        R4["返回缓存值"]
        R5["repository.find() + 写入缓存"]
        R1 --> R2
        R2 --> R3
        R3 -- "是" --> R4
        R3 -- "否" --> R5
    end

    subgraph "写路径"
        W1["save(namespace, key, value)"]
        W2["repository.save(namespace, key, value)"]
        W3["⚠️ 不写缓存"]
        W1 --> W2 --> W3
    end

    subgraph "删路径"
        D1["delete(namespace, key)"]
        D2["cache.remove()"]
        D3["repository.delete(namespace, key)"]
        D1 --> D2 --> D3
    end

四、核心 API 草图

ConfigStorage<T extends ConfigDefinition>

// default 方法实现，组合 repository + cache + namespace
interface ConfigStorage<T extends ConfigDefinition> {
    void save(String key, T value);
    void save(String namespace, String key, T value);
    Optional<T> load(String namespace, String key);
    void delete(String namespace, String key);
    boolean exists(String namespace, String key);
    Set<String> listKeys(String namespace);
    Map<String, T> loadAll(String namespace);
    Map<String, T> loadAll(String namespace, Collection<String> keys);
    void saveAll(String namespace, Map<String, T> configs);
    void deleteAll(String namespace);
    void deleteKeys(String namespace, Collection<String> keys);
    <E> E loadAttachment(String namespace, String key, Function<T, Object> creator);
    String cacheKey(String namespace, String key);  // namespace:key
}

WatchableConfigStorage<T>

interface WatchableConfigStorage<T> extends ConfigStorage<T> {
    WatchHandle watch(String namespace, String key, ConfigChangeListener<T> listener);
    WatchHandle watchNamespace(String namespace, ConfigChangeListener<T> listener);
    WatchHandle watchKeys(String namespace, Collection<String> keys, ConfigChangeListener<T> listener);
    void unwatch(WatchHandle handle);
}

ConfigRepository<T extends ConfigDefinition>

// 持久化后端 SPI，只管原始数据存取，不管缓存和命名空间逻辑
interface ConfigRepository<T extends ConfigDefinition> {
    T find(String namespace, String key);
    List<T> list(String namespace);
    void save(String namespace, String key, T value);       // upsert 模式
    void delete(String namespace, String key);              // 删单个
    void delete(String namespace);                          // 删整个命名空间
}

ConfigDefinition

@Data
class ConfigDefinition {
    String id;           // 主键
    String namespace;    // 逻辑命名空间
    String key;          // 配置键
    String description;  // 描述
    String engine;       // 引擎标识（如 "rules"、"flows"）
    String text;         // 配置文本内容
    Map<String, Object> extra;  // 扩展字段
    int version;         // 版本号
    Object attachment;   // 懒加载附加对象（运行时填充）
}

五、包结构与核心类

easy-store/
└── io/matrix/store/
    ├── ConfigStorage.java                      # 顶层接口（default 方法组合 Cache + Repository + Namespace）
    ├── WatchableConfigStorage.java             # 监听能力接口
    ├── AbstractConfigStorage.java               # 抽象基类（持有 repository + cache 字段）
    ├── AbstractWatchableConfigStorage.java      # 监听实现（维护 watchers + fireConfigChangeEvent）
    ├── DefaultConfigStorage.java                # 唯一具体实现
    ├── ConfigDefinition.java                   # 通用配置 POJO
    ├── ConfigChangeEvent.java                  # 变更事件（oldValue/newValue/ChangeType）
    ├── ConfigChangeListener.java               # @FunctionalInterface 监听回调
    ├── WatchHandle.java                        # 监听句柄接口
    └── repository/                             # 持久化后端 SPI + 实现
        ├── ConfigRepository.java               # SPI 接口
        ├── InMemoryRepository.java             # 后端：ConcurrentHashMap
        ├── FileRepository.java                 # 后端：YAML 文件（非泛型）
        └── RedisRepository.java                # 后端：RedisTemplate（半成品）

依赖：

六、缓存与监听机制

6.1 缓存策略：Cache-Aside（写跳过缓存）

flowchart LR
    subgraph "读路径（Cache-Aside）"
        L1["load()"] --> L2["cache.computeIfAbsent()"]
        L2 -->|"miss"| L3["repository.find()"]
        L3 --> L4["写入 cache<br/>key=namespace:key"]
    end
    subgraph "写路径（跳过缓存）"
        W1["save()"] --> W2["repository.save()"]
        W2 -->|"⚠️ 不写 cache"| W3["如果该 key 之前<br/>被缓存过则读旧值"]
    end
    subgraph "删路径（清缓存）"
        D1["delete()"] --> D2["cache.remove()"]
        D2 --> D3["repository.delete()"]
    end

设计权衡：save 跳缓存避免缓存穿透，但 save 后立即 load 可能读到旧值（如果之前已缓存过）。delete 先清缓存确保读不到脏数据。

潜在问题：save 不清缓存的行为可能不是故意的设计，更像是 bug——save 之后应该先清缓存（或直接写新值到缓存），否则 save → load 的读写序列会返回旧数据。

6.2 监听机制：被动触发

AbstractWatchableConfigStorage.fireConfigChangeEvent() 是 protected 方法，遍历 watchers 并判断 shouldNotify：

protected void fireConfigChangeEvent(
    String namespace, String key,
    T oldValue, T newValue,
    ConfigChangeEvent.ChangeType changeType) {
    // 遍历 watchers，shouldNotify 判断后调 listener.onChange(event)
}

当前状态：DefaultConfigStorage 没有重写 save/delete 来调 fireConfigChangeEvent，所以监听机制目前不会自动触发。实际使用有两条路：

1. 子类继承 DefaultConfigStorage，在 save/delete 之后手动调 fire
2. 用 AOP 拦截 ConfigStorage 的 save/delete 方法触发

6.3 Watcher 结构

Map<String, WatcherInfo<T>> watchers
    ├── key = "namespace"  → WatcherInfo(namespace, null, isNamespaceWatch=true)
    ├── key = "namespace:key" → WatcherInfo(namespace, key, isNamespaceWatch=false)
    └── ...

CompositeWatchHandle 是 watchKeys 批量监听时的返回值，cancel 时递归取消所有子 WatchHandle。

七、关键设计决策

八、已知问题与待改进