DDK Projection Starter

让读模型跟着写模型走：搜索索引、统计表、列表页用的宽表。数据变了，在同一个事务里做一个标记；提交之后刷新读模型，失败重试，需要时整体重建。0.8.0 起。

text

应用服务（一个事务）
  保存聚合
  projections.markDirty("order-search", orderId)   → 待刷新表里多一行
提交
  └─ 后台线程：OrderSearchProjection.refresh(orderId)
       读取订单现在的状态 → 写进读模型 → 删掉那一行

能做什么

能力	说明
标记随事务提交	标记和业务修改一起提交、一起回滚。数据改了而读模型没跟上的情况，不会因为进程崩溃而出现
提交后立即刷新	另有定时清扫，处理遗留的标记和到期的重试
合并重复标记	同一条数据在刷新之前被标记多次，只刷新一次
不丢变更	一条数据在刷新期间又变了，标记会留下来再刷新一次
失败重试	等待时间逐次翻倍，有上限，失败原因记在表里
重建	把写模型里的每一条都刷新一遍，中途重启不影响结果

刷新，而不是应用

DDK 的聚合是按状态保存的，没有可以重放的事件日志；事件还可能乱序到达。所以投影没有做成「把每个事件的变化应用到读模型上」，而是：数据变了之后，用写模型里的当前状态重新生成读模型里的这一条。

java

@Component
@RequiredArgsConstructor
public class OrderSearchProjection implements Projection {

    private final OrderRepository orders;
    private final OrderSearchIndex index;      // 应用自己定义的、通往读模型存储的端口
    private final JdbcTemplate jdbc;

    @Override
    public String name() {
        return "order-search";
    }

    @Override
    public void refresh(String id) {
        orders.find(OrderId.of(Long.valueOf(id)))
                .ifPresentOrElse(order -> index.save(OrderDocument.from(order)), () -> index.delete(id));
    }

    @Override
    public void forEachId(Consumer<String> ids) {          // 只有重建才需要
        jdbc.query("SELECT id FROM t_order", rs -> {
            ids.accept(rs.getString(1));
        });
    }
}

这样做的三个好处：

重复执行没有副作用。 refresh 是覆盖式的，重试一次结果不变。但同一条数据的两次刷新互相重叠是另一回事，见下面的多个实例。
不依赖事件的先后。 同一个聚合的两个事件无论谁先到，做的都是「读取当前状态」。
重建走的是同一段代码。 没有事件历史可以重放，也不需要。

事件只负责说明哪一条变了：

java

@EventListener
public void on(OrderPaidEvent event) {
    projections.markDirty("order-search", event.orderId());
}

用普通的 @EventListener，标记才会加入发出事件的那个事务。

引入方式

xml

<dependency>
    <groupId>com.domaindrivenkit</groupId>
    <artifactId>ddk-projection-starter</artifactId>
</dependency>

应用里至少有一个 Projection Bean，并且有 JdbcTemplate 和事务管理器时生效。

各种情况下的行为

情况	行为
事务提交	标记随之提交，随即开始刷新
事务回滚	没有标记，不刷新
提交之后进程崩溃	标记还在表里，下一次清扫时刷新
刷新之前同一条数据被标记多次	刷新一次
刷新期间这条数据又变了	标记不会被这次刷新删掉，随后再刷新一次
`refresh` 抛出异常	推迟重试，等待时间翻倍直到上限
多个实例	同一时刻一条标记只被一个 worker 领取，见下文
worker 刷新到一半停顿或崩溃	领取在 `lease-time` 后过期，由其他 worker 接手

在事务之外调用 markDirty，标记立即写入并立即刷新。

多个实例

所有实例共用一张待刷新表。每条标记带两种身份：

列	什么时候变	回答什么问题
`generation`	这条数据再次被标记时；每次生成新的随机值，永不复用	我开始处理之后，数据又变过没有？
`claim_token`、`claimed_until`	worker 领取标记时，带一个租约	我现在还是处理它的那个 worker 吗？

text

领取标记（只领取未被领取、或租约已过期的）
refresh(id)                       读源数据，写读模型
DELETE ... WHERE generation = 我领取时的代次 AND claim_token = 我的令牌
  删到了   -> 领取以来没有变化，我写入的就是最新状态
  没删到   -> 数据又变了，或者租约过期被别人接手：
              我写入的可能是旧状态，所以重新标记这条数据

一个 worker 停顿太久、失去租约之后醒来，仍然可能把旧快照写进读模型。但它藏不住：它的确认落空，这条数据会再刷新一次。一条数据在表里没有标记时，读模型里的它等于最后一次刷新时写模型的状态。 删除也一样：旧写入把已删除的数据带回来，随后的刷新会再次删掉它。

在那次补刷之前，读模型可能短暂地显示旧快照。要求读模型任何时刻都不回退时，在 refresh 里按源版本做条件写入：关系库用 WHERE source_version < ?，Elasticsearch 用 version_type=external_gte。DDK 的聚合自带这个版本（AggregateRoot.version()）。

lease-time 要长于一次正常刷新的耗时。设短了不会出错，只是浪费：慢的刷新会被接手，这条数据刷新两次。

0.8 及以前的说法是错的

0.8 的文档写的是「多个实例各自清扫，同一条数据可能被刷新不止一次，依据是 refresh 可以重复执行」。重复执行确实无害，但重叠不是：实例 A 读到 v1 后停顿，实例 B 读 v2、写入、删除标记，A 醒来写回 v1，结果读模型停在 v1 而表里没有任何标记。0.9 起用上面的代次和租约修复了这个问题。

多租户

开启 ddk-mybatis-starter 的多租户后，每条 SQL 都要从 OperatorContext 取租户，而后台 worker 自己没有。所以待刷新表把租户和标记一起存下来：

markDirty 记录当前 OperatorContext 的租户。主键是 (projection, tenant_id, entity_id)，只在租户内唯一的标识在两个租户下是两条标记。
worker 刷新有租户的标记时，用 OperatorContext.runAs 以系统操作者 ddk-projection（RefreshWorker.OPERATOR_ID）加这个租户的身份执行，结束后恢复线程原来的上下文。租户取自表里的那一行，所以进程重启或换一个实例接手，都以原来的租户刷新。
没有租户的标记照旧不设置操作者。在多租户应用里，这意味着 refresh 的第一条查询就失败，按退避重试，原因写进 last_error。starter 不猜测租户，也不为了让刷新通过而放宽多租户。
rebuild 在调用方的上下文里枚举标识，所以重建的是调用方所在的租户。要重建全部租户，在 OperatorContext.runAs 里逐个租户调用。pending 统计全部租户。

text

markDirty("orders", 1)   在租户 A 的请求里   →  (orders, A, 1)
markDirty("orders", 1)   在租户 B 的请求里   →  (orders, B, 1)
worker                                         →  runAs(ddk-projection@A, refresh("1"))
                                                  runAs(ddk-projection@B, refresh("1"))

重建

java

long marked = projections.rebuild("order-search");   // 标记每一个标识，标记写完就返回
long left = projections.pending("order-search");     // 还有多少条在等

重建只做标记，刷新由后台线程完成。没处理完的标记留在表里，进程重启后继续。重建不会删除读模型里多出来的数据（写模型里已经没有的那些），需要清理时先清空读模型再重建。

运维

一条标记一直失败不会挡住别的标记：它按翻倍的间隔重试，上限是 max-retry-delay，失败次数和最近一次的错误留在表里。找到并恢复它们用 Projections：

java

List<FailedRefresh> failing = projections.failures("order-search", 20);   // 租户、标识、失败次数、最近的错误、下次重试时间
int retried = projections.retryFailed("order-search");                    // 修好原因之后立即重试，不等退避

retryFailed 保留失败次数，再失败时退避从原来的位置继续。整个过程不需要手工改表。

classpath 上有 Micrometer（例如引入了 Spring Boot Actuator）时，starter 输出：

指标	类型	含义
`ddk.projection.refresh{projection, outcome}`	Timer	每次刷新及耗时。`outcome` 是 `success`、`repeated`（期间数据又变了或被别人接手，已重新标记）或 `failed`
`ddk.projection.pending{projection}`	Gauge	等待刷新的标记数，包括等待重试的
`ddk.projection.failing{projection}`	Gauge	至少失败过一次、等待重试的标记数
`ddk.projection.oldest.age{projection}`	Gauge，秒	已到期的标记里等得最久的那条等了多久。持续上升说明 worker 跟不上或卡住了

这几个 gauge 在采集时查询待刷新表。建议的告警：oldest.age 超过几个清扫间隔；failing 大于零的时间超过 max-retry-delay；failed 的速率上升。

配置项

配置项	默认值	说明
`ddk.projection.enabled`	`true`	是否注册 `Projections` 并处理待刷新的标记
`ddk.projection.table`	`ddk_projection_pending`	待刷新表的表名
`ddk.projection.initialize-schema`	`true`	启动时自动建表；用迁移工具管理表结构时关掉
`ddk.projection.sweep-interval`	`30s`	清扫间隔，处理其他实例留下的、重启前没处理完的、到了重试时间的标记
`ddk.projection.batch-size`	`100`	每次从表里取多少条
`ddk.projection.retry-delay`	`5s`	第一次重试前的等待时间，之后每失败一次翻倍
`ddk.projection.max-retry-delay`	`5m`	重试等待时间的上限
`ddk.projection.lease-time`	`2m`	一条标记被领取后归一个 worker 处理多久，过期后其他 worker 可以接手

取值不合法时启动失败：batch-size 小于 1、任一时长为零或负数、retry-delay 大于 max-retry-delay。

sql

CREATE TABLE ddk_projection_pending (
    projection    VARCHAR(100) NOT NULL,
    tenant_id     VARCHAR(64)  DEFAULT '' NOT NULL,
    entity_id     VARCHAR(200) NOT NULL,
    version       BIGINT       NOT NULL,
    generation    VARCHAR(36),
    claim_token   VARCHAR(36),
    claimed_until TIMESTAMP,
    attempts      INT          NOT NULL,
    available_at  TIMESTAMP    NOT NULL,
    last_error    VARCHAR(500),
    PRIMARY KEY (projection, tenant_id, entity_id)
);
CREATE INDEX ddk_projection_pending_due ON ddk_projection_pending (available_at);

这个索引给清扫用：清扫按 available_at 找到期的标记。已有的表缺这个索引时，initialize-schema=true 会在启动时补上；为 false 时只记录一条带语句的警告，因为缺索引只会让清扫变慢，不影响正确性。它的效果没有做过基准测试。

从 0.8 升级

0.8 建的表没有 tenant_id、generation、claim_token、claimed_until，主键也不含租户。各数据库改主键的写法不同，所以 starter 改为复制重建，用的都是 MySQL、PostgreSQL、H2 通用的语句：

sql

CREATE TABLE ddk_projection_pending_upgrade AS SELECT * FROM ddk_projection_pending;
DROP TABLE ddk_projection_pending;
CREATE TABLE ddk_projection_pending (...);          -- 上面的表结构
CREATE INDEX ddk_projection_pending_due ON ddk_projection_pending (available_at);
INSERT INTO ddk_projection_pending (projection, tenant_id, entity_id, version, generation, attempts, available_at, last_error)
    SELECT projection, '', entity_id, version, CONCAT('upgraded-', version), attempts, available_at, last_error
    FROM ddk_projection_pending_upgrade;
DROP TABLE ddk_projection_pending_upgrade;

initialize-schema=true（默认）：启动时执行这些语句。已有的标记保留原来的状态，算作没有租户。
initialize-schema=false：启动失败，并打印这些语句，放进迁移脚本。表完全不存在时同样启动失败，而不是等到第一次刷新才报错。
复制重建不是原子的（MySQL 的每条 DDL 单独提交）。中途停下会留下 ddk_projection_pending_upgrade，下次启动接着做完：旧表还完整就丢掉副本重来，否则建好新表、把新表里还没有的标记从副本搬回来。initialize-schema=false 时启动失败并打印这些收尾语句。

升级时先停掉旧实例。复制期间 0.8 实例写入的标记会丢失；0.8 的 worker 按旧协议工作，也可能留下旧数据。升级期间旧实例还在运行的话，升级后重建一次投影。tenant_id 的默认值保证这段时间里旧实例插入标记不会失败。

不用 RENAME：PostgreSQL 重命名表之后，旧主键的索引名仍是 <表名>_pkey，新表建主键时会撞名。

不适用的场景 / 已知问题

由其他服务的事件拼出来的读模型：它没法从本应用自己的写模型重新生成。这种情况用 Event Starter 消费事件，在消费方里更新读模型。
写入量很大的场景：每个实例一个后台线程，逐条刷新，没有并行，也没有对读模型存储的批量写入。
不管读模型存在哪里：starter 不和 Elasticsearch 或其他存储打交道，那是 refresh 里的事。参考应用 ddk-mall 里有关系库和 Elasticsearch 两个实现可以对照。Elasticsearch 专用的支持（索引管理、重建时切换别名）还没有做。
一直失败的标记会一直重试，最多每隔 max-retry-delay 一次，没有「放弃」状态。用 Projections.failures 或 ddk.projection.failing 指标找到它们，见运维。
读模型有延迟：它在提交之后才刷新。写完立刻读，可能读到旧数据。
验证范围：同一套集成测试在 H2、MySQL 8.4、PostgreSQL 17 上运行，其中包括用闩锁和可控时钟编排的并发场景：两次刷新重叠、租约过期后被接手、旧确认碰到新标记、删除后被旧写入复活。

DDK Projection Starter ​

能做什么 ​

刷新，而不是应用 ​

引入方式 ​

各种情况下的行为 ​

多个实例 ​

多租户 ​

重建 ​

运维 ​

配置项 ​

从 0.8 升级 ​

不适用的场景 / 已知问题 ​

DDK Projection Starter

能做什么

刷新，而不是应用

引入方式

各种情况下的行为

多个实例

多租户

重建

运维

配置项

从 0.8 升级

不适用的场景 / 已知问题