领域建模战术：解密事件回放与读模型视图重构规范

在采用**事件溯源（Event Sourcing，简称 ES）**配合 **CQRS（读写分离）** 架构的领域建模设计中，写端（Write Side）保存着只读的领域事件流，而读端（Read Side）则通过投影（Projection）处理器维护着面向复杂查询的关系视图。

然而，当业务迭代需要对查询视图进行升级重构时（例如：读端数据库表需要新增一个根据历史交易记录计算出的「信用等级」字段；或者原有的读端数据库损坏需要数据恢复），我们应该如何处理？在传统的单体架构中，这通常需要停机进行复杂的历史数据清洗（ETL）。但在事件溯源架构下，我们能通过**事件回放（Event Replay）**，从头播放全部历史事件流，在完全零停机的情况下重新构建出全新的读模型视图。

本文将系统拆解事件回放的战略定位与多视图重建设计、零停机双视图平滑切换拓扑、以及 Java 事件回放处理器的落地开发规范。

一、 核心对比：传统 ETL 数据清洗 vs. ES 事件回放重构

两种历史数据重构方案在开发复杂度、系统停机时间以及数据准确度上差异显著：

二、 读模型蓝绿重建与零停机双视图平滑切换拓扑

在不中断线上服务的情况下，通过将事件流向新版本视图表重放、并利用路由切换实现零停机发布的拓扑如下：

      [ 外部客户端高并发查询请求 (GET /accounts) ]
                           │
                           ▼ 【 1. 阶段一：当前正常路由 】
              ┌──────────────────────────┐
              │ 路由网关指向: 旧版视图表 │ ──> (T_ACCOUNT_VIEW_V1)
              └──────────────────────────┘
                           │
      【 2. 阶段二：后台开启事件回放重建 (蓝绿发布) 】
      - 创建新物理表: T_ACCOUNT_VIEW_V2
      - 启动 EventReplayProcessor，从 Event Store 第 0 号事件开始拉取
      - 将事件源源不断转换为 V2 格式写入 T_ACCOUNT_VIEW_V2
                           │
                           ▼ 3. 回放速度赶上主时间轴 (Catch-Up 阶段)
      【 4. 阶段三：双向同步写入 (Dual Write / Catch-Up Sync) 】
      - 线上实时发生的新事件，同时向 V1 投影器与 V2 投影器投递，保持 V2 实时对齐
                           │
                           ▼ 5. 版本路由无缝切换
              ┌──────────────────────────┐
              │ 路由网关切换: 新版视图表 │ ──> (T_ACCOUNT_VIEW_V2)
              └──────────────────────────┘
                           │
      【 6. 销毁或废弃 T_ACCOUNT_VIEW_V1，发布圆满结束 】

三、 代码实战：在 Java 中编写事件回放与投影重组引擎

以下代码展示了如何 design 一个事件回放处理器，以游标分批拉取只读事件流，并向全新的 V2 视图表中注入历史状态：

1. 领域事件与新读视图实体（Domain & Read Model）

package com.company.sales.domain.event;

import java.io.Serializable;

public class MoneyDepositedEvent implements Serializable {
    private final String accountId;
    private final double amount;
    private final long timestamp;

    public MoneyDepositedEvent(String accountId, double amount, long timestamp) {
        this.accountId = accountId;
        this.amount = amount;
        this.timestamp = timestamp;
    }

    public String getAccountId() { return accountId; }
    public double getAmount() { return amount; }
    public long getTimestamp() { return timestamp; }
}

2. 基础设施层：高可用事件回放重建引擎（Infra Layer）

package com.company.infra.cqrs.rebuild;

import com.company.sales.domain.event.MoneyDepositedEvent;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import java.util.List;

@Service
public class EventReplayProjectionRebuilder {

    private final JdbcTemplate jdbcTemplate;
    private static final int BATCH_SIZE = 1000; // 分批拉取大小

    public EventReplayProjectionRebuilder(JdbcTemplate template) {
        this.jdbcTemplate = template;
    }

    /**
     * 核心规约：执行全新 V2 视图的后台只读回放重建
     */
    public void rebuildV2Projection() {
        System.out.println("[Rebuilder] 开始初始化 V2 关系视图表...");
        // 1. 初始化新物理表，加入最新业务需要的信用度分（credit_score）字段
        jdbcTemplate.execute("DROP TABLE IF EXISTS t_account_view_v2");
        jdbcTemplate.execute("CREATE TABLE t_account_view_v2 (account_id VARCHAR(50) PRIMARY KEY, balance DOUBLE, credit_score INT)");

        long lastEventId = 0; // 事件流水游标
        boolean hasMore = true;

        System.out.println("[Rebuilder] 启动事件流水游标扫描，执行 Event Replay...");
        while (hasMore) {
            // 2. 游标分批拉取历史只读事件，防止 OOM 崩溃
            String selectSql = "SELECT event_id, payload, event_type FROM t_event_store WHERE event_id > ? ORDER BY event_id ASC LIMIT ?";
            List<RawDbEvent> rawEvents = jdbcTemplate.query(
                selectSql, 
                new BeanPropertyRowMapper<>(RawDbEvent.class), 
                lastEventId, 
                BATCH_SIZE
            );

            if (rawEvents.isEmpty()) {
                hasMore = false;
                break;
            }

            // 3. 执行内存反序列化并转换为投影更新
            for (RawDbEvent raw : rawEvents) {
                MoneyDepositedEvent event = deserialize(raw.getPayload());
                applyToV2Projection(event);
                lastEventId = raw.getEventId(); // 推进游标
            }
        }
        System.out.println("[Rebuilder] 历史事件回放完毕，新视图 T_ACCOUNT_VIEW_V2 重建完成！");
    }

    private void applyToV2Projection(MoneyDepositedEvent event) {
        // 在新视图 V2 中累加余额，并动态推算附加字段：每充值 100 元加 1 个信用分
        int additionalCredit = (int) (event.getAmount() / 100);
        
        String updateSql = "UPDATE t_account_view_v2 SET balance = balance + ?, credit_score = credit_score + ? WHERE account_id = ?";
        int updated = jdbcTemplate.update(updateSql, event.getAmount(), additionalCredit, event.getAccountId());
        
        if (updated == 0) {
            String insertSql = "INSERT INTO t_account_view_v2 (account_id, balance, credit_score) VALUES (?, ?, ?)";
            jdbcTemplate.update(insertSql, event.getAccountId(), event.getAmount(), additionalCredit);
        }
    }

    private MoneyDepositedEvent deserialize(String payload) {
        // 模拟 JSON 转换
        String[] parts = payload.replace("{", "").replace("}", "").split(",");
        String accountId = parts[0].split(":")[1].replace(""", "");
        double amount = Double.parseDouble(parts[1].split(":")[1]);
        return new MoneyDepositedEvent(accountId, amount, System.currentTimeMillis());
    }

    public static class RawDbEvent {
        private long eventId;
        private String payload;
        private String eventType;

        public long getEventId() { return eventId; }
        public void setEventId(long eventId) { this.eventId = eventId; }
        public String getPayload() { return payload; }
        public void setPayload(String payload) { this.payload = payload; }
        public String getEventType() { return eventType; }
        public void setEventType(String eventType) { this.eventType = eventType; }
    }
}
```