java

9.1 eventd v3 架构

eventd v3 完整架构设计：Event 领域模型与统一事件格式、Kafka Streams 清洗管道（去重/标准化/过滤）的拓扑设计、Flink DataStream 实时计算（窗口聚合/CEP/指标计算）的作业设计、Redis 缓存层与实时面板的推拉模式。

Leeting Yan 2026-11-01 5 分钟阅读 2134 字

eventd v3 的定位

eventd v3 是贯穿卷三的实时数据管道，承接卷二 eventd v2 的事件输出：

eventd v2 (WebFlux + R2DBC)
    └── 业务事件 → Kafka Topic "eventd-raw"
                           ↓
    Kafka Streams (eventd-cleansing)
        ├── 去重（基于 eventId）
        ├── 格式标准化（统一字段名、数据类型）
        ├── 异常过滤（丢弃非法数据）
        └── 写入 Topic "eventd-cleaned"
                           ↓
    Flink DataStream (eventd-analytics)
        ├── 1 分钟滚动窗口：每分钟每用户订单量
        ├── 5 分钟滑动窗口：近5分钟平均延迟
        ├── CEP：连续3次登录失败告警
        └── 结果写入 Redis
                           ↓
    实时面板 (eventd-panel, WebFlux SSE)
        └── 浏览器实时刷新指标

模块	技术	输入	输出
eventd-cleansing	Kafka Streams Topology	eventd-raw	eventd-cleaned
eventd-analytics	Flink DataStream	eventd-cleaned	Redis
eventd-panel	WebFlux + SSE	Redis	浏览器

统一事件模型

eventd v3 的所有模块共享同一 Event record（9.1 之前讨论的是 v2 的 POJO 形态，v3 改成不可变 record）：

package com.plume.eventd.shared;

import java.util.Map;
import java.util.UUID;

public record Event(
    String eventId,        // 全局唯一，用于去重
    String type,           // 事件类型：ORDER/LOGIN/REFUND/LOGOUT
    int version,           // 事件格式版本
    String source,         // 来源系统：eventd-v2
    long eventTime,        // Event Time（毫秒级 Unix 时间戳）
    String userId,         // 用户标识
    double amount,         // 金额（ORDER/REFUND 有效）
    String symbol,         // 商品/品类标识
    int status,            // 状态码：0=成功，1=失败
    String ipAddress,      // IP 地址（风控用）
    String device,         // 设备类型
    Map<String, String> metadata  // 扩展字段
) {
    public Event {
        if (eventId == null) eventId = UUID.randomUUID().toString();
        if (version == 0) version = 1;
        if (source == null) source = "eventd-v2";
        if (metadata == null) metadata = Map.of();
    }

    public boolean isValid() {
        if (eventId == null || eventId.isEmpty()) return false;
        if (type == null || type.isEmpty()) return false;
        if (eventTime <= 0) return false;
        if (userId == null || userId.isEmpty()) return false;
        return true;
    }

    public static Event.Builder builder() {
        return new Event.Builder();
    }

    // 构造器只有 12 个参数的全量版本，测试、生产方如果逐字段拼装会很难读，
    // 所以给一个 builder。字段默认值与构造器里的补全规则保持一致
    public static class Builder {
        private String eventId;
        private String type;
        private int version = 1;
        private String source = "eventd-v2";
        private long eventTime;
        private String userId;
        private double amount;
        private String symbol;
        private int status;
        private String ipAddress;
        private String device;
        private Map<String, String> metadata = Map.of();

        public Builder eventId(String eventId) { this.eventId = eventId; return this; }
        public Builder type(String type) { this.type = type; return this; }
        public Builder version(int version) { this.version = version; return this; }
        public Builder source(String source) { this.source = source; return this; }
        public Builder eventTime(long eventTime) { this.eventTime = eventTime; return this; }
        public Builder userId(String userId) { this.userId = userId; return this; }
        public Builder amount(double amount) { this.amount = amount; return this; }
        public Builder symbol(String symbol) { this.symbol = symbol; return this; }
        public Builder status(int status) { this.status = status; return this; }
        public Builder ipAddress(String ipAddress) { this.ipAddress = ipAddress; return this; }
        public Builder device(String device) { this.device = device; return this; }
        public Builder metadata(Map<String, String> metadata) {
            this.metadata = metadata;
            return this;
        }

        public Event build() {
            return new Event(eventId, type, version, source, eventTime, userId,
                amount, symbol, status, ipAddress, device, metadata);
        }
    }
}

字段	必填	验证
eventId	✅	非空且全局唯一；格式由生产方约定（UUID 是常见选择，验收/测试里也常用可读前缀）
type	✅	enum: ORDER, LOGIN, LOGOUT, REFUND
eventTime	✅	> 0
userId	✅	非空
amount	否	缺失按 0；负数不丢整条事件，在 `normalizeEvent` 里截断为 0（ORDER/REFUND 的零值需由业务规则另行判断）

模块划分

eventd-cleansing（Kafka Streams）

Input: eventd-raw (JSON 字符串)
    │
    ├── Deserialize（JSON → Event）
    ├── Validate（字段检查、格式校验）
    ├── Deduplicate（eventId 去重，KeyValueState）
    ├── Normalize（字段名统一、类型转换）
    └── Serialize（Event → JSON）
    │
Output: eventd-cleaned (JSON 字符串)

核心拓扑：

StreamsBuilder builder = new StreamsBuilder();

// 从 eventd-raw 读取
KStream<String, String> rawStream = builder.stream("eventd-raw",
    Consumed.with(Serdes.String(), Serdes.String()));

// 清洗管道
KStream<String, String> cleanedStream = rawStream
    .mapValues(this::deserialize)
    .filter((key, event) -> event.isValid())
    .processValues(() -> new DeduplicateProcessor("dedup-store"), "dedup-store")
    .mapValues(this::normalize)
    .mapValues(this::serialize);

// 写入 eventd-cleaned
cleanedStream.to("eventd-cleaned");

eventd-analytics（Flink DataStream）

Input: eventd-cleaned (Kafka Source)
    │
    ├── Assign Timestamp + Watermark
    ├── KeyBy(userId + symbol)
    ├── Window (Tumbling/Sliding/Session)
    ├── Aggregate (Count/Sum/Avg)
    └── Sink to Redis

主作业：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 时间语义不在这里设置：由下面 fromSource 的 WatermarkStrategy 决定
env.enableCheckpointing(30000);
env.setStateBackend(new EmbeddedRocksDBStateBackend());

// Kafka Source
KafkaSource<Event> source = KafkaSource.<Event>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("eventd-cleaned")
    .setGroupId("eventd-analytics")
    .setStartingOffsets(OffsetsInitializer.latest())
    .setValueOnlyDeserializer(new EventDeserializationSchema())
    .build();

DataStream<Event> stream = env.fromSource(source,
    WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
        .withIdleness(Duration.ofMinutes(5))
        .withTimestampAssigner((event, ts) -> event.eventTime()),
    "kafka-source"
);

// 窗口聚合：每分钟每用户订单量
stream.keyBy(e -> e.userId())
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .aggregate(new OrderCountAggregate())
    .addSink(new RedisMetricSink("order-count"));

// CEP：风控规则
Pattern<Event, ?> pattern = Pattern.<Event>begin("fails")
    .where(e -> e.type().equals("LOGIN") && e.status() != 0)
    .timesOrMore(3)
    .within(Duration.ofMinutes(5));

CEP.pattern(stream.filter(e -> e.type().equals("LOGIN")).keyBy(Event::userId), pattern)
    .process(new AlertHandler())
    .addSink(new RedisAlertSink());

env.execute("eventd-v3 analytics");

eventd-panel（WebFlux SSE）

浏览器 → HTTP GET /api/stream/metrics
              ↓
    ServerController (WebFlux)
              ↓
    Redis Reactive → Lettuce
              ↓
    flux.map(json → ServerSentEvent)
              ↓
    SSE 持续推送

@RestController
public class PanelController {
    private final ReactiveRedisTemplate<String, String> redisTemplate;

    @GetMapping(value = "/api/stream/metrics", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> streamMetrics() {
        // 频道名要和 9.3 里 RedisMetricSink 的 publish 完全一致。
        // 需要通配时才用 PatternTopic.of("eventd:*")——glob 匹配的是整个频道名，
        // 写成 "metrics:*" 匹配不到 "eventd:metrics"
        return redisTemplate.listenTo(ChannelTopic.of("eventd:metrics"))
            .map(message -> ServerSentEvent.builder(message.getMessage())
                .id(UUID.randomUUID().toString())
                .build())
            // 用 sample 做刷新节流：每个 500ms 窗口只发最后一条。
            // delayElements 不是节流，它把每条消息都排队延后，消息一快就积压
            .sample(Duration.ofMillis(500));
    }
}

数据流拓扑图

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  eventd-raw  │────▶│    Kafka     │────▶│   eventd-    │
│   (input)    │     │   Streams    │     │  cleaned     │
└──────────────┘     └──────────────┘     └──────────────┘
                                                │
                                          ┌─────┴─────┐
                                          ▼           ▼
                                    ┌─────────┐  ┌──────────┐
                                    │  Window │  │   CEP    │
                                    │Aggregate│  │  Alert   │
                                    └────┬────┘  └────┬─────┘
                                         │            │
                                         ▼            ▼
                                   ┌──────────┐  ┌──────────┐
                                   │  Redis   │  │  Redis   │
                                   │ metrics  │  │  alerts  │
                                   └────┬─────┘  └────┬─────┘
                                        │             │
                                        ▼             ▼
                                  ┌────────────────────────────┐
                                  │   eventd-panel (WebFlux)   │
                                  │      SSE 实时推送           │
                                  └────────────────────────────┘

配置项汇总

配置	值	说明
Kafka raw Topic	`eventd-raw`	v2 输出
Kafka cleaned Topic	`eventd-cleaned`	清洗后
Flink parallelism	4	开发机 4 核
Checkpoint interval	30s	容错
Watermark delay	10s	乱序容忍
Window size	1min	分钟级指标
Redis key prefix	`eventd:metrics:`	指标存储
SSE delay	500ms	面板刷新频率

与卷二的衔接

eventd v2 → v3 的衔接点：

卷二输出	v3 消费	格式
`POST /api/orders`	写入 eventd-raw	Event JSON
`POST /api/login`	写入 eventd-raw	Event JSON
`POST /api/refund`	写入 eventd-raw	Event JSON

eventd v2 的 OrderService.createOrder() 增加事件发送：

// 在卷二的 OrderService 基础上增加
public Mono<Order> createOrder(OrderRequest request) {
    return orderRepo.save(request.toOrder())
        .flatMap(order -> eventPublisher.publish(
            Event.builder()
                .type("ORDER")
                .userId(order.userId())
                .amount(order.amount())
                .symbol(order.symbol())
                .eventTime(System.currentTimeMillis())
                .build()
        ).thenReturn(order));
}

小结

模块	技术	核心职责
eventd-cleansing	Kafka Streams	去重、标准化、过滤
eventd-analytics	Flink DataStream	窗口聚合、CEP、指标计算
eventd-panel	WebFlux SSE	实时推送
eventd-shared	Java Record	统一事件模型

下一节 9.2 Kafka Streams 清洗管道：去重、格式标准化、异常过滤的完整实现。

继续阅读

探索更多技术文章

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

全部文章返回首页