java

12.1 eventd v3 全链路集成

eventd v3 全链路集成：从 eventd v2 Kafka Producer 到 Kafka Streams 清洗、再到 Flink DataStream 聚合、Redis Sink 指标缓存与 SSE 面板的完整端到端可运行流程与配置调优。

Leeting Yan 2026-11-01 7 分钟阅读 3164 字

全链路数据流

eventd v3 的完整数据链路：

┌─────────────────────────────────────────────────────────────────────┐
│                         eventd v2 (卷二)                            │
│  WebFlux + R2DBC                                                    │
│                        ↓ 业务事件                                   │
│                     Kafka Producer                                  │
└─────────────────────────────────────────────────────────────────────┘
                           │
                           ▼
                    ┌──────────────┐
                    │ eventd-raw   │
                    │ (Kafka Topic)│
                    └──────┬───────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
          ▼                ▼                ▼
  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐
  │ Deserializer │  │   Validate   │  │ Dead Letter  │
  │  JSON→Event  │  │ 字段检查     │  │ 无效数据     │
  └──────────────┘  └──────────────┘  └──────────────┘
          │
          ▼
  ┌──────────────┐
  │ Deduplicate  │  KeyValueStore<eventId, Boolean>
  │  Based on    │  TTL 7 days
  │  eventId     │
  └──────┬───────┘
         │
         ▼
  ┌──────────────┐
  │ Normalize    │  类型转大写、金额非负、补齐字段
  └──────┬───────┘
         │
         ▼
  ┌──────────────┐
  │ Serialize    │  JSON
  └──────┬───────┘
         │
         ▼
┌─────────────────────────────────────────────────────────────────────┐
│                    eventd-cleansing (Kafka Streams)                 │
└─────────────────────────────────────────────────────────────────────┘
                           │
                           ▼
                    ┌──────────────┐
                    │ eventd-clean │
                    │ ed (Kafka)   │
                    └──────┬───────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────────────┐
│                    eventd-analytics (Flink)                         │
│                                                                     │
│  ─── Source ───────────────────────────────────────────             │
│  KafkaSource + WatermarkStrategy(10s乱序容忍)                       │
│                                                                     │
│  ─── Branch 1 ──────────────────────────────────────                │
│  keyBy(userId) + window(1min Tumbling) + order count                │
│                                                                     │
│  ─── Branch 2 ──────────────────────────────────────                │
│  keyBy(常量) + window(5min Sliding) + avg latency（全局维度）        │
│                                                                     │
│  ─── Branch 3 ──────────────────────────────────────                │
│  keyBy(symbol) + window(1min Tumbling) + hot product                │
│                                                                     │
│  ─── Branch 4 ──────────────────────────────────────                │
│  CEP: 连续3次登录失败 → alert                                       │
│                                                                     │
│  ─── Sink ───────────────────────────────────────────               │
│  RedisMetricSink (TTL 5min) + RedisAlertSink (TTL 1h)               │
└─────────────────────────────────────────────────────────────────────┘
                           │
                           ▼
                    ┌──────────────┐
                    │ Redis        │
                    │              │
                    │ eventd:      │
                    │ metrics:*    │
                    │ eventd:      │
                    │ alerts       │
                    └──────┬───────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────────────┐
│                     eventd-panel (WebFlux)                          │
│                                                                     │
│  /api/stream/metrics  → SSE (Redis pub/sub)                         │
│  /api/stream/alerts   → SSE (Redis pub/sub)                         │
│  /api/health          → 监控端点                                    │
└─────────────────────────────────────────────────────────────────────┘
                           │
                           ▼
                    ┌──────────────┐
                    │ 浏览器面板   │
                    └──────────────┘

集成配置

Kafka Topic 规划

Kafka 没有「topic 配置文件」这个概念，下面的表只是建 Topic 的规格清单；真实创建要落到 kafka-topics.sh 或运维脚本里（见下节启动顺序）。

Topic	分区	副本	说明
`eventd-raw`	12	1	v2 写入的原始事件，决定 Kafka Streams 的有效并行度上限
`eventd-cleaned`	12	1	Kafka Streams 清洗后的标准事件，Flink Source 读它
`eventd-dead-letter`	1	1	校验失败的原消息，串行保存便于人工排查

分区数一旦少于上游需要的并行度，后面只能扩分区重新分布；副本数要 ≤ broker 数，单机 Docker 环境只能设 1。

eventd-cleansing 配置

eventd-cleansing 用 Spring Kafka 托管 Streams，真正的配置键是下面这些（processing.guarantee 的值是 exactly_once_v2）：

# cleansing/application.yaml
spring:
  kafka:
    # 本地 mvn 启动用 localhost；compose 里用环境变量覆盖成 kafka:9092
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    streams:
      application-id: eventd-cleansing
      properties:
        processing.guarantee: exactly_once_v2
        # EOS 下必须严格小于 transaction.timeout.ms（Streams 默认 10s）
        commit.interval.ms: 5000
        statestore.cache.max.bytes: 10485760   # 3.x 新名，旧名 cache.max.bytes.buffering 已废弃
        num.stream.threads: 4

eventd:
  cleansing:
    dedup-ttl-hours: 168

eventd-analytics 配置

Flink 的配置键有固定前缀，混写自定义层级不会被识别：

# analytics/flink-conf.yaml
parallelism.default: 4

execution.checkpointing.interval: 30s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.min-pause: 10s
execution.checkpointing.timeout: 60s

state.backend.type: rocksdb
state.checkpoints.dir: file:///var/flink/checkpoints

Source 并行度取 Kafka 分区数、Sink 并行度取 4，这类差异在代码里用 setParallelism 指定，不写在集群配置里。Watermark 的 10 秒乱序容忍和空闲判定属于 WatermarkStrategy，同样在代码里。

env.fromSource(kafkaSource, WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
        .withIdleness(Duration.ofMinutes(5)),
        "eventd-cleaned")
    .setParallelism(12);

eventd-panel 配置

# panel/application.yaml
spring:
  data:                 # Spring Boot 3 起 Redis 配置在 spring.data.redis 下
    redis:
      host: ${REDIS_HOST:localhost}
      port: 6379

sse:
  metrics-interval-ms: 500
  alert-interval-ms: 1000
  max-connections: 1000

启动顺序

# 1. 启动 Kafka 集群
kafka-server-start.sh config/server.properties

# 2. 创建 Topic
kafka-topics.sh --bootstrap-server localhost:9092 \
    --create --topic eventd-raw --partitions 12 --replication-factor 1
kafka-topics.sh --bootstrap-server localhost:9092 \
    --create --topic eventd-cleaned --partitions 12 --replication-factor 1
kafka-topics.sh --bootstrap-server localhost:9092 \
    --create --topic eventd-dead-letter --partitions 1 --replication-factor 1

# 3. 启动 eventd-cleansing
cd eventd-v3/eventd-cleansing
mvn compile exec:java -Dexec.mainClass="com.plume.eventd.CleansingApplication"

# 4. 启动 eventd-analytics
cd eventd-v3/eventd-analytics
mvn compile exec:java -Dexec.mainClass="com.plume.eventd.AnalyticsJob"

# 5. 启动 eventd-panel
cd eventd-v3/eventd-panel
mvn spring-boot:run

# 6. 验证面板
curl -s http://localhost:8083/api/health

mvn compile exec:java 需要项目里声明 exec-maven-plugin；前面两步也可以用 java -cp target/classes:$(cat cp.txt) com.plume.eventd.CleansingApplication 这类方式启动。Kafka Streams 是库，进程起来就是一个 Streams 客户端，没有「提交作业」的步骤。

端到端测试

下面这段是集成冒烟测试，需要 Kafka、Flink、Redis 都已启动。它最容易踩的坑不是断言写错，而是「等不到窗口结果」：单条事件的时间戳落在某个 1 分钟滚动窗口内，而 Watermark 只有来了一条时间戳更晚的事件才会推进，窗口才可能关闭。

@Test
void testFullPipeline() throws Exception {
    String bootstrap = System.getenv().getOrDefault(
            "KAFKA_BOOTSTRAP_SERVERS", "localhost:9092");
    KafkaProducer<String, String> producer = new KafkaProducer<>(
        Map.of(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap,
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class
        )
    );

    // 1700000000 秒不是 60 的整数倍，先向下取整到整分钟，窗口才是 [base, base+60s)
    long base = 1_699_999_980_000L;
    // 六参 event(id, type, userId, amount, symbol, ts) 工厂沿用 9.3 的定义
    Event first = event("e-1", "ORDER", "alice", 100.0, "BTC", base + 1_000L);
    // 第二条把 Watermark 推过窗口结束（10s 乱序容忍 + 1s 余量）。
    // 必须和第一条同 key：不同 key 会进不同分区，Flink 的 Watermark 取各分区最小值，
    // alice 所在分区一直不推进的话窗口照样不关。用 LOGIN 是为了让它被订单分支过滤掉
    Event pumper = event("e-pump", "LOGIN", "alice", 0.0, "BTC", base + 60_000L + 11_000L);

    producer.send(new ProducerRecord<>("eventd-raw", first.userId(), JsonMapper.toJson(first)));
    producer.send(new ProducerRecord<>("eventd-raw", pumper.userId(), JsonMapper.toJson(pumper)));
    producer.flush();

    // 轮询而不是 sleep：清洗 + 聚合 + 写入链路耗时不定
    // base 已对齐到整分钟，直接就是窗口起点
    String key = "eventd:metrics:order-count:" + base + ":alice";
    RedisCommands<String, String> redis = RedisClient
        .create("redis://redis:6379").connect().sync();

    String value = null;
    long deadline = System.currentTimeMillis() + 30_000;
    while (value == null && System.currentTimeMillis() < deadline) {
        value = redis.get(key);
        if (value == null) Thread.sleep(200);
    }
    assertNotNull(value, "30 秒内没有等到 " + key);
}

三点约定：

断言的是具体窗口 key（metric:windowStart:dimensionValue），不是 :latest 之类的指针——指针由哪条记录最后写入是不确定的。
Redis key 里的时间戳用事件时间窗口起点，与 Sink 的格式化规则保持一致；两边规则不一致时测试会永远等不到。
9.3 的 RedisMetricSink 写入的是 JSON（如 {"count":1}），所以这里能直接取到值；如果 Sink 改成了紧凑串，断言前必须先按同样的格式解码。

还有一个容易忽略的前提：Flink 的 Watermark 取各 Kafka 分区的最小值。测试 topic 若有多分区且只有部分分区有数据，闲置分区会一直压住 Watermark，窗口永远不关。要么测试建单分区 topic，要么在测试作业里把 withIdleness 调到 1 分钟；生产代码用的是 5 分钟，对一次几十秒的冒烟测试来说等不起。

故障排查

现象	排查	解决
面板无数据	Redis key 存在？Flink 运行？	检查 Flink Web UI
数据重复	去重状态 TTL？Kafka commit？	检查 dedup-store 状态
延迟高	Watermark 推进？反压？	检查 Web UI backpressure
告警不触发	CEP pattern 匹配？	检查事件时间偏移

小结

阶段	模块	输出	验证点
生产	eventd v2	eventd-raw	Kafka 收到消息
清洗	eventd-cleansing	eventd-cleaned	去重 + 标准化
聚合	eventd-analytics	Redis	窗口统计 + 告警
展示	eventd-panel	浏览器	SSE 推送

下一节 12.2 部署与运维：Standalone / Docker / 配置管理。

继续阅读

探索更多技术文章

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

全部文章返回首页