books

《Java 消息中间件·事件驱动架构》11.2 Topic 命名与生命周期

Kafka Topic 命名规范与生命周期治理：domain.event.action.vN 格式、CREATING→ACTIVE→DEPRECATED→RETAINED→DELETED 五态机，以及孤儿主题自动清理。Java 模拟命名校验与非法状态转换拦截。

Leeting Yan 6 分钟阅读 2790 字

命名规范不是风格指南

主题的命名看起来是一件小事，但当平台里有几百个主题时，命名不规范的代价是：运维不知道从哪找日志、开发者不知道消费哪个主题、血缘工具无法自动识别域归属。命名规范在事件平台里不是"代码风格"，而是基础设施的入口。

命名规范必须满足三个要求：

  1. 可解析：从名字里能提取 domain、event type、action。
  2. 可排序：同一 domain 的事件能按名字分组。
  3. 可版本化：带 vN 后缀，区分同一事件的不同 schema 版本。

domain.event.action.vN 格式

本书采用的命名格式是四层点分：

domain.event.action.vN
段示例含义校验规则
domainlibrary业务域全小写，与事件目录中的 domain 一致
eventloan事件对象名词，聚合根名
actioncreated动作过去时，描述状态变化
versionv1版本号v 前缀 + 正整数，双写阶段靠它区分

四个反例：

反例问题
loanCreated不是点分，无法按域分组
library.loan.created缺少版本号，schema 升级时无法双写
library.Loan.Created.v1大写，Kafka 特性允许但跨系统迁移时大小写会丢失
library.loan.create.v1action 不是过去时，暗示是命令不是事件

五态生命周期

一个主题从诞生到销毁，至少经历五个状态：

状态说明触发条件
CREATING正在创建中工单通过 INIT 校验
ACTIVE可读写工单到达 CREATED
DEPRECATED只读不写新版本主题已上线，旧主题停止接收写入
RETAINED归档保留保留期（如 90 天）内可读，仅用于重放
DELETED已删除保留期结束，数据清除

状态转换有严格的合法路径：

CREATING → ACTIVE → DEPRECATED → RETAINED → DELETED

非法转换举例：ACTIVE → DELETED（缺少 DEPRECATED 过渡，消费者没时间迁移）、RETAINED → ACTIVE（归档后不能再激活，必须新建）。

纯 Java 命名与生命周期模拟

下面的 TopicLifecycleManager 包含命名规范校验、五态机转换、以及孤儿主题检测（90 天无消费标记为孤儿）。

// TopicLifecycleManager.java — Topic 命名校验、生命周期管理与孤儿检测

import java.time.*;
import java.util.*;

public class TopicLifecycleManager {

    enum State { CREATING, ACTIVE, DEPRECATED, RETAINED, DELETED }

    record Topic(String name, State state, Instant lastConsumedAt,
                 Instant createdAt, Instant deprecatedAt) {

        boolean namingValid() {
            return name.matches("^[a-z]+\\.[a-z]+\\.[a-z]+\\.v[0-9]+$");
        }
    }

    static final Map<String, Long> ORPHAN_THRESHOLD_DAYS = Map.of(
        "library", 90L,
        "member",  60L,
        "order",   30L,
        "payment", 30L
    );

    static State transition(Topic topic, State target) {
        State current = topic.state();
        // 合法转换
        if (current == State.CREATING && target == State.ACTIVE) return target;
        if (current == State.ACTIVE && target == State.DEPRECATED) return target;
        if (current == State.DEPRECATED && target == State.RETAINED) return target;
        if (current == State.RETAINED && target == State.DELETED) return target;

        System.out.println("  [ILLEGAL] 无法从 " + current + " 转换到 " + target);
        return current; // 转换失败，保持原状态
    }

    static boolean isOrphan(Topic topic, Instant now) {
        if (topic.state() != State.ACTIVE && topic.state() != State.DEPRECATED) {
            return false; // 只有活跃或弃用的主题可能成为孤儿
        }
        String domain = topic.name().split("\\.")[0];
        long threshold = ORPHAN_THRESHOLD_DAYS.getOrDefault(domain, 90L);
        long daysUnused = Duration.between(topic.lastConsumedAt(), now).toDays();
        return daysUnused > threshold;
    }

    static void demoLifecycle() {
        System.out.println("=== Topic 生命周期管理 ===\n");

        // 创建主题
        Topic t = new Topic("library.loan.created.v1", State.CREATING,
            Instant.now(), Instant.now(), null);
        System.out.println("创建: " + t.name() + " -> " + t.state());
        System.out.println("  命名规范: " + (t.namingValid() ? "PASS" : "FAIL"));

        // 激活
        t = new Topic(t.name(), transition(t, State.ACTIVE), t.lastConsumedAt(),
            t.createdAt(), t.deprecatedAt());
        System.out.println("转正: " + t.name() + " -> " + t.state());

        // 弃用
        t = new Topic(t.name(), transition(t, State.DEPRECATED), t.lastConsumedAt(),
            t.createdAt(), Instant.now());
        System.out.println("弃用: " + t.name() + " -> " + t.state());

        // 归档
        t = new Topic(t.name(), transition(t, State.RETAINED), t.lastConsumedAt(),
            t.createdAt(), t.deprecatedAt());
        System.out.println("归档: " + t.name() + " -> " + t.state());

        // 删除
        t = new Topic(t.name(), transition(t, State.DELETED), t.lastConsumedAt(),
            t.createdAt(), t.deprecatedAt());
        System.out.println("删除: " + t.name() + " -> " + t.state());

        System.out.println();
    }

    static void demoIllegalTransition() {
        System.out.println("=== 非法状态转换 ===\n");

        Topic active = new Topic("member.profile.updated.v1", State.ACTIVE,
            Instant.now(), Instant.now(), null);
        System.out.println("试图: ACTIVE -> DELETED");
        transition(active, State.DELETED);

        Topic retained = new Topic("order.payment.confirmed.v2", State.RETAINED,
            Instant.now(), Instant.now(), Instant.now());
        System.out.println("试图: RETAINED -> ACTIVE");
        transition(retained, State.ACTIVE);

        System.out.println();
    }

    static void demoNamingCheck() {
        System.out.println("=== 命名规范检查 ===\n");
        String[] cases = {
            "library.loan.created.v1",   // 通过
            "loanCreated",               // 失败：点分
            "library.loan.created",      // 失败：缺少版本
            "library.Loan.Created.v1",   // 失败：大写
            "library.loan.create.v1",    // 失败：非过去时
            "order.invoice.generated.v2" // 通过
        };
        for (String c : cases) {
            Topic t = new Topic(c, State.CREATING, Instant.now(), Instant.now(), null);
            System.out.printf("  %-40s -> %s%n", c, t.namingValid() ? "PASS" : "FAIL");
        }
        System.out.println();
    }

    static void demoOrphanDetection() {
        System.out.println("=== 孤儿主题检测 ===\n");
        Instant now = Instant.now();

        // library 主题：91天未消费 -> 孤儿
        Topic libOrphan = new Topic("library.reservation.cancelled.v1",
            State.ACTIVE, now.minus(Duration.ofDays(91)),
            now.minus(Duration.ofDays(100)), null);
        System.out.println(libOrphan.name() + " 最后消费=" + libOrphan.lastConsumedAt()
            + " -> 孤儿=" + isOrphan(libOrphan, now));

        // member 主题：61天未消费 -> 孤儿（member 阈值 60 天）
        Topic memOrphan = new Topic("member.points.expired.v1",
            State.ACTIVE, now.minus(Duration.ofDays(61)),
            now.minus(Duration.ofDays(70)), null);
        System.out.println(memOrphan.name() + " 最后消费=" + memOrphan.lastConsumedAt()
            + " -> 孤儿=" + isOrphan(memOrphan, now));

        // order 主题：29天未消费 -> 非孤儿（order 阈值 30 天）
        Topic ordActive = new Topic("order.shipment.delivered.v1",
            State.ACTIVE, now.minus(Duration.ofDays(29)),
            now.minus(Duration.ofDays(40)), null);
        System.out.println(ordActive.name() + " 最后消费=" + ordActive.lastConsumedAt()
            + " -> 孤儿=" + isOrphan(ordActive, now));

        // 已弃用主题：即使很久没消费也不算孤儿（需要等待保留期）
        Topic deprecated = new Topic("payment.refund.processed.v1",
            State.DEPRECATED, now.minus(Duration.ofDays(120)),
            now.minus(Duration.ofDays(130)), now.minus(Duration.ofDays(100)));
        System.out.println(deprecated.name() + " 状态=DEPRECATED 最后消费="
            + deprecated.lastConsumedAt() + " -> 孤儿=" + isOrphan(deprecated, now));
    }

    public static void main(String[] args) {
        demoLifecycle();
        demoIllegalTransition();
        demoNamingCheck();
        demoOrphanDetection();
    }
}

本机实测（Temurin 25.0.5+8 / 2026-10-29）：
javac --release 21 TopicLifecycleManager.java && java TopicLifecycleManager

=== Topic 生命周期管理 ===

创建: library.loan.created.v1 -> CREATING
  命名规范: PASS
转正: library.loan.created.v1 -> ACTIVE
弃用: library.loan.created.v1 -> DEPRECATED
归档: library.loan.created.v1 -> RETAINED
删除: library.loan.created.v1 -> DELETED

=== 非法状态转换 ===

试图: ACTIVE -> DELETED
  [ILLEGAL] 无法从 ACTIVE 转换到 DELETED
试图: RETAINED -> ACTIVE
  [ILLEGAL] 无法从 RETAINED 转换到 ACTIVE

=== 命名规范检查 ===

  library.loan.created.v1                  -> PASS
  loanCreated                              -> FAIL
  library.loan.created                     -> FAIL
  library.Loan.Created.v1                  -> FAIL
  library.loan.create.v1                   -> FAIL
  order.invoice.generated.v2               -> PASS

=== 孤儿主题检测 ===

library.reservation.cancelled.v1 最后消费=2026-07-30... -> 孤儿=true
member.points.expired.v1 最后消费=2026-08-29... -> 孤儿=true
order.shipment.delivered.v1 最后消费=2026-09-30... -> 孤儿=false
payment.refund.processed.v1 状态=DEPRECATED 最后消费=... -> 孤儿=false

命名规范中的 vN 不是装饰

很多人把版本后缀当成"以后再说"的装饰，实际上它是事件治理的基础设施：

场景没有 vN 的后果有 vN 的方案
双写过渡新旧版本必须共用同一个主题名，消费者被迫同时解析两个 schemav1 和 v2 是两个独立主题，可以分别管理
消费者迁移不知道哪个 schema 版本是当前消费的目标消费者配置里写死 v1，迁移到 v2 时只改一行
保留期管理无法为不同版本设置不同保留策略library.loan.created.v1 设 30 天，v2 设 90 天

命名规范中的 vN 不是装饰，而是防止双写阶段同名冲突的第一道防线。

孤儿主题清理流程

孤儿主题不是垃圾——它们可能还在被某个离线消费者消费，或者历史数据仍有合规保留要求。清理流程必须分阶段：

  1. 标记：检测到孤儿后先标记，不删除，通知业务团队。
  2. 确认：给业务团队 7 天确认，确认无消费则进入下一状态。
  3. DEPRECATED → RETAINED：先停止写入，再进入只读保留期。
  4. RETAINED → DELETED：保留期结束，数据清除。

即使已经标记为孤儿，如果业务团队突然说"还有人在用"，平台也不能直接删除。这是治理与运营的边界：治理工具通知、运营决策删除。

不同域的不同阈值

代码里的 ORPHAN_THRESHOLD_DAYS 按域分别设置：

域阈值理由
library90 天教育场景，借阅周期较长，寒暑假期间消费稀疏
member60 天会员体系，活跃周期中等
order/payment30 天交易场景，消息时效性强，30 天未消费即异常

阈值的设定不是技术决策，而是业务决策。同一个平台的不同域可以有不同的阈值，甚至同一域内的不同事件也可以有不同的阈值（如 payment.fraud.alert 可能只需要 7 天）。

本节小结

  • 命名规范是可解析、可排序、可版本化的基础设施，不是风格指南。
  • 命名格式：domain.event.action.vN，四层点分，全小写，action 用过去时。
  • 五态生命周期：CREATING → ACTIVE → DEPRECATED → RETAINED → DELETED，非法转换被拦截。
  • 命名规范中的 vN 不是装饰，而是双写阶段防止同名冲突的第一道防线。
  • 孤儿主题检测按域分别设阈值，标记后须经业务确认再清理。
  • 清理流程分四步：标记→确认→DEPRECATED→RETAINED→DELETED。
  • 非法状态转换必须在代码层面拦截，不能依赖"不会有人这么做"的假设。
  • DEPRECATED 状态的主题不算孤儿——它需要等待保留期，不是消费活跃度决定。

阅读导航：上一节 11.1 事件平台自助化 ｜ 下一节 11.3 配额与成本分摊

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收