World Map · 工程方法论蒸馏

E5 - 处理基础：转换 + 确定性 + 分布式时间 ​

3 步处理流程 + 4 种转换 + 确定性流 + 分布式时间同步——事件驱动微服务的"运行时基础"。

一句话 ​

流处理 = 消费 + 处理 + 生产 3 步循环；确定性 = 同事件序列 → 同结果（无论何时重放）；分布式时间 = NTP 同步 + 事件时间 + 处理时间 + 水位线。

这本书在这一层说了什么 ​

本层融合 EDM Ch 5（Transformations，15KB / 2412 词）+ Ch 6（Synchronizing Distributed Timestamps，38KB / 6118 词）。

Bellemare Ch 5 关键论断：

"Most event-driven microservices follow, at a minimum, the same three steps:

  1. Consume an event from an input event stream.
  2. Process that event.
  3. Produce any necessary output events."

核心论点 5 个：

  1. The 3-Step Processing Loop — 标准处理循环
  2. Composing Stateless Topologies — 4 种 Transformations
  3. Branching and Merging Streams — 流分支与合并
  4. Synchronizing Distributed Timestamps — 分布式时间同步
  5. Event Scheduling and Deterministic Processing — 确定性调度

是什么 ​

1. 3 步处理循环（标准模式） ​

Bellemare Ch 5 伪代码：

java
Consumer consumerClient = new consumerClient(consumerGroupName, ...);
Producer producerClient = new producerClient(...);

while(true) {
    InputEvent event = consumerClient.pollOneEvent(inputEventStream);
    OutputEvent output = processEvent(event);
    producerClient.produceEventToStream(outputEventStream, output);

    // At-least-once processing
    consumerClient.commitOffsets();
}

关键洞察：

  • At-least-once：消费后先 commit offsets（可能重复处理）
  • processEvent = 业务入口——业务逻辑全在这里
  • 持续循环——不停歇服务（除非优雅关闭）

2. 4 种 Transformations（变换） ​

Bellemare 关键分类：

Transformation输入输出关键
Filter1 event0 or 1 event满足条件才传播
Map1 event1 event改 key/value（改 key 需 repartition）
MapValue1 event1 event只改 value（无需 repartition）
Custom1 event0+ events自定义逻辑（查状态 / 同步外部）

Filter 例子：

python
# 保留金额 > 10.0 的事件
def filter_large_events(event):
    if event.value.amount > 10.0:
        return event
    return None  # 过滤掉

Map 例子：

python
# 改 key（需要 repartition）
def map_key(event):
    return Event(
        key=event.value.customer_id,  # 改 key
        value=event.value
    )

关键洞察：改 key = 改分区，可能破坏顺序。

3. 流分支与合并 ​

分支（Branch）：

python
# 按国家路由到不同流
def branch_by_country(event):
    if event.value.country == "US":
        produce('orders-us', event)
    elif event.value.country == "EU":
        produce('orders-eu', event)
    else:
        produce('orders-other', event)

# 错误处理：死信流
def handle_with_dlq(event):
    try:
        process(event)
    except Exception as e:
        produce('dead-letter', event, error=str(e))

合并（Merge）：

流 1 ─┐
流 2 ─┼─→ 合并处理 → 输出流
流 3 ─┘

Bellemare 关键警告：

"If you do end up merging event streams, define a new unified schema representing the resulting merged events."

关键：合并流必须定义统一 Schema，否则消费者无法解析。

4. 分布式时间同步（NTP） ​

Bellemare 关键洞察：

"A fundamental limitation of physics is that two independent systems cannot be guaranteed to have precisely the same system-clock time."

NTP 同步精度：

同步场景漂移
局域网 + NTP几毫秒（15 分钟后）
局域网 + 高频 NTP1ms 或更少
跨互联网±100ms

NTP 失效场景：

  • 网络中断
  • 配置错误
  • 多租户 VM 干扰

Bellemare 关键论断：

"For the vast majority of business cases, frequent synchronization to NTP servers can provide sufficient consistency for system event time."

关键：业务时间不需要精确同步——NTP 够用。

5. 时间类型 2 种 ​

Bellemare 关键概念：

事件时间（Event Time）：
  - 事件实际发生的时间
  - 写入事件的字段中
  
处理时间（Processing Time）：
  - 服务处理事件的时间
  - 服务本地时钟

二者差异：

  • 乱序事件（事件时间 < 处理时间）
  • 迟到事件（处理时间 - 事件时间 > 阈值）

6. 确定性处理（Deterministic Processing） ​

Bellemare 关键定义：

"Deterministic processing requires that events be processed consistently, such that the results can be reproduced at a later date."

确定性 = 同输入 → 同输出，无论何时处理。

例子（银行账户）：

反例（确定性破坏）：
  按 round-robin 处理多 partition
  $10 存款 (事件时间 10:00) + $25 取款 (事件时间 09:55) + $20 存款 (事件时间 10:05)
  → 顺序处理 → 负余额 → 罚款 ❌

正确（确定性）：
  按事件时间戳排序处理
  $25 取款 (09:55) + $10 存款 (10:00) + $20 存款 (10:05)
  → 顺序处理 → 余额 $5 ✅

Bellemare 关键论断：

"The processing order of events must be interleaved based on the event time provided in the record, regardless of which input partition it comes from, to ensure correct results."

7. 事件调度策略 ​

Bellemare 关键论断：

"The most common event-scheduling implementation selects and dispatches the next events based on their event time..."

调度策略：

  • 单 partition → 按 offset 顺序
  • 多 partition → 按 event time 交错
  • 跨流 → 按 event time 全局排序

为什么 ​

为什么需要 At-Least-Once？ ​

反例（At-Most-Once）：

  • 消费后立即 commit offsets
  • 处理失败 → 事件丢失
  • 业务状态错误

At-Least-Once：

  • 处理成功后 commit
  • 处理失败 → 重试
  • 可能重复处理 → 必须幂等

根因：业务数据不能丢，重复可接受（业务层去重）。

为什么需要确定性处理？ ​

根因 1：可重放

  • 业务故障 → 重放历史事件 → 恢复状态
  • 重放必须产生相同结果

根因 2：可调试

  • 业务疑问 → 重放到特定时间点
  • 重放结果必须一致

根因 3：可测试

  • 测试用历史数据
  • 测试结果必须可重现

为什么改 key 要 repartition？ ​

根因：Kafka 等系统保证单 partition 内有序，不保证跨 partition。

反例：

  • 事件 A（key=user1）在 partition 1
  • 事件 B（key=user1）改 key=user2 → repartition 到 partition 2
  • 处理顺序：A、B 不再有先后保证 → 业务错乱

修复：尽量不改 key（用 MapValue 代替 Map）；改 key 时充分理解后果。

为什么需要 NTP 同步？ ​

根因：业务事件有"何时发生"的语义。

  • 金融交易 → 哪笔先发生？
  • 订单履约 → 时间敏感
  • 审计 → 事件顺序

没有 NTP：事件时间戳不可信 → 业务语义破坏。

怎么做 ​

Step 1：标准处理循环实现 ​

python
from kafka import KafkaConsumer, KafkaProducer
import json

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='kafka:9092',
    group_id='order-processor',
    enable_auto_commit=False  # 手动 commit
)
producer = KafkaProducer(
    bootstrap_servers='kafka:9092',
    value_serializer=lambda v: json.dumps(v).encode()
)

while True:
    for message in consumer:
        try:
            # 处理
            input_event = json.loads(message.value)
            output_event = process_event(input_event)

            # 生产（必须成功）
            producer.send('processed-orders', value=output_event).get()

            # Commit（处理成功后）
            consumer.commit()
        except Exception as e:
            # 不 commit → 下次重试
            log.error(f"处理失败: {e}")
            continue

Step 2：实现幂等（避免重复处理） ​

python
# 用 event_id 去重
processed_ids = set()

def process_event(event):
    if event['id'] in processed_ids:
        return None  # 跳过重复
    processed_ids.add(event['id'])
    # 业务逻辑...
    return processed_event

Step 3：按事件时间排序处理 ​

python
from sortedcontainers import SortedList

# 缓存 + 排序
pending_events = SortedList(key=lambda e: e.event_time)

for message in consumer:
    event = parse(message.value)
    pending_events.add(event)

    # 按时间顺序处理就绪事件
    while pending_events and pending_events[0].event_time <= watermark:
        e = pending_events.pop(0)
        output = process_event(e)
        producer.send('output', output)
        consumer.commit()

Step 4：水位线（Watermark）机制 ​

python
# 水位线 = "事件时间 < X 的事件应该都已到达"
class WatermarkTracker:
    def __init__(self, max_lateness=60):
        self.max_lateness = max_lateness  # 60 秒迟到容忍
    
    def update(self, event_time):
        self.current_watermark = event_time - timedelta(seconds=self.max_lateness)
    
    def should_process(self, event):
        return event.event_time >= self.current_watermark

Step 5：NTP 监控 ​

bash
# 检查 NTP 偏移
ntpq -p

# 监控指标
chronyc tracking

什么时候用 / 什么时候不用 ​

用流处理的 2 大场景 ​

场景关键判断
事件流持续输入流处理 = 持续消费 + 处理 + 生产
需要确定性 + 可重放按事件时间排序 + 水位线

不用流处理（或简化）的 2 大场景 ​

场景原因
简单事件 → 单输出直接同步处理即可，无需复杂拓扑
数据量小 + 无重放需求批处理更简单

看不见的地方 ​

失效模式 1：非确定性导致重放不一致 ​

症状：流处理含 Date.now() / Random → 重放结果不同 → 业务错误。

根因：处理函数有副作用。

修复：所有副作用通过确定性函数表达，或明确标注"非确定性边界"。

失效模式 2：错误处理吃掉失败 ​

症状：try-catch + log → 失败事件丢失 → 业务状态错。

根因：未区分"可恢复"与"不可恢复"错误。

修复：可恢复错误重试；不可恢复错误死信队列。

失效模式 3：顺序假设被破坏 ​

症状：按事件顺序假设处理 → 分区乱序时业务错。

根因：未按 Key 分区。

修复：必须按 Key 分区（保证同 Key 顺序）。

失效模式 4：重复处理未做幂等 ​

症状：消费者收到重复事件 → 重复扣库存 → 数据错。

根因：未做幂等。

修复：消费逻辑必须幂等（用 eventId 去重）。

失效模式 5：水位线失准 ​

症状：业务要求实时但水位线设了 60 秒容忍 → 实际延迟 60 秒。

根因：水位线设置不合理。

修复：根据业务 SLA 调整水位线（容忍越低 → 越易丢失迟到事件）。

相关节点 ​

骨架 A — 世界观 ​

节点关联解释
wonder-03-emergence-time强事件时间 = "时"的物理形式——按时间排序处理就是按"时"涌现
wonder-02-emergence-action中处理循环 = "涌现作为"——持续消费/处理/生产的"作为"循环

骨架 B — 工程实践 ​

节点关联解释
value-stream强流处理 = 价值流的连续处理——不停歇处理每条价值
quality-assurance中确定性 = 质量保障——重放测试的基础

状态 ​

E5 概念递进落地完成（融合原书 ch05 15KB + ch06 38KB）。

  • 5 个核心论点（3 步循环 / 4 变换 / 流分支 / NTP / 确定性调度）
  • 5 步怎么做（含完整伪代码 + 水位线机制）
  • 5 个失效模式
  • 骨架 A/B 完整映射

从名家方法论与工程化思路中蒸馏出自己的工程体系。

© World Map