05与 Flink 流计算的同构

前一章把 Actor 放到 Go CSP 与 Rust 所有权旁边做了「语言级」对照。本章换一个视角:把 Actor 放到 Apache Flink 这类流计算框架旁边,做「架构级」对照。因为 Actor 一旦要扩展到分布式、高吞吐、有状态、要容错的场景,它撞上的难题与 Flink 的算子(operator)几乎如出一辙——而 Flink 二十年沉淀下来的工程答卷,恰好能反过来点拨我们如何设计 Actor 系统。

这并非牵强附会。Flink 的算子本质上就是一个个有状态、按 key 分区、以数据流相连、支持 checkpoint 与故障恢复的处理单元——把「算子」换成「Actor」,把「数据流」换成「消息」,两套体系便严丝合缝地重叠了。更进一步说,这层重叠不是措辞上的巧合:两边其实都在回答同一个更底层的问题——一个长期运行、有状态的处理单元,要如何在不可靠的分布式环境里既保证正确性又不崩溃。谁先给出经过大规模生产验证的答案,谁的经验就值得另一边直接搬过来用。

四组核心同构

🔑 设计钥匙

这四组同构不是巧合,而是同一个问题在两个领域里长出的同一套答案:一旦系统「有状态 + 分布式 + 要容错」,你迟早会需要「按 key 隔离状态」「端到端背压」「状态独立于执行体」「有上限的重启」这四件事。Flink 已经替你把这四件事趟平了——这意味着设计 Actor 系统时,你可以直接借用 Flink 成熟的答案,而不必从零发明。

下面逐组展开,指出本笔记 demo 里的落点,以及更重要的一点——这层同构为什么值得当真,而不只是一个好看的类比。

Keyed State ≈ 按地址分区的 Actor

Flink 里,keyBy(sessionId) 会把同一个 key 的所有数据路由到同一个算子实例:这个实例在整个作业运行期间独占这个 key 的状态,Flink 的运行时保证「同 key 必同实例、同实例内消息严格串行处理」。正因为有「串行处理」这个前提,不同 key 的状态才互不可见,算子内部读写自己的 keyed state 完全不需要加锁——锁在这里根本没有存在的必要,因为从设计上就不存在两个线程同时碰同一份状态的时刻。水平扩展也随之而来:key 按哈希分进若干个 key group,key group 再分布到不同的并行子任务上,加并行度就是把 key group 摊得更开,各 key 的状态始终各自独立、互不打扰。

这和 Actor 是同一件事:本 demo 里的 ManagerActor 就是那个「按 key 路由」的分发器,它按 sessionID 把请求路由到对应的 SessionActor,每个会话的 token 历史、进度都是该 Actor 的独占状态。mgr.sessions[m.ID] 这个 map 就是 Flink 的 keyed-state 目录,SpawnChild 相当于「为一个新 key 惰性创建算子实例」——不是所有 key 都会同时活跃,惰性创建避免了为潜在的百万会话预先分配百万个 Actor。下面这段交互动画演示了这个过程:三个会话 A / B / C 的消息并发涌向 ManagerActor,被按 key 分流到各自的 SessionActor,每个 Actor 对私有计数器做裸的自增操作,三份状态互不干扰、无需任何锁。

// ManagerActor 持有 sessionID -> 会话 PID 的私有映射。单线程访问，无需锁。
type ManagerActor struct {
  sessions map[string]*actor.PID[SessionMsg]
}

这层同构值得当真的原因,不只是「概念对得上」,而是它把 Flink 用生产事故换来的一条经验直接白送给你:分区键的选择决定了系统的可扩展性上限。Flink 社区吃过热 key 的亏——如果某个 key 的数据量远超其它 key,对应的并行子任务就会成为整条流水线的瓶颈,加再多并行度也无济于事,因为那一个 key 永远只能落在一个实例上。这个教训原样适用于 Actor:某个热点 SessionActor 的 mailbox 会持续堆积,而你没法靠「多开几个 Actor」分担——按 key 路由的规则决定了这个 key 只能去这一个地址。知道了这一点,你在设计阶段就能提前避坑:要么把重负载再按子 key 拆分,要么在业务层面限制单 key 的负载上限——这些都是 Flink 处理数据倾斜时的标准手段,直接照搬即可,不必自己从头摸索。

Credit 背压 ≈ 有界 mailbox

Flink 早期用 TCP 滑动窗口做背压:下游处理不过来,TCP 连接的接收缓冲区被填满,内核自然不再从上游读取数据,压力经由网络协议栈反向传导。这个办法简单,却有一个致命缺陷——Flink 的网络层会在同一条 TCP 连接上复用多个逻辑数据流,一旦某个逻辑流量堆积,就会连累共享这条连接的其它逻辑流。后来的 credit-based flow control 把背压做成了显式的、按逻辑通道独立核算的机制:下游告诉上游「我这条通道还有多少 buffer 余量(credit)」,上游必须持有 credit 才能发送,credit 耗尽就停——每条逻辑通道各算各的账,互不牵连,背压端到端地精确传导到源头,而不是在内存里囤积数据直到打爆。

有界 mailbox 就是这个机制的微缩版,而且天生没有「多路复用打架」的问题——因为一个 mailbox 从创建起就只属于一个 Actor,不存在「多个逻辑流共享一条物理通道」这回事,credit-based 要额外解决的问题在 Actor 里根本不会发生。本 demo 里背压链条清晰可见:慢客户端 → SSE 订阅 channel 满 → SessionActor 在 emit 阻塞 → 它的 mailbox 堆积 → worker 的 TellBlocking 被迫等待 → 生成降速。mailbox 底层的 post 是非阻塞投递:队列满或已关闭就立即返回失败,把决定权交还给调用方;postBlocking 则会一直等到有空位或 mailbox 关闭为止。

// post 非阻塞投递。返回 false 表示 mailbox 已满（背压忙信号）或已关闭。
func (m *mailbox[T]) post(e envelope[T]) bool {
  // 先判关闭，避免向已关闭 mailbox 投递。
  select {
  case <-m.done:
    return false
  default:
  }
  select {
  case m.user <- e:
    return true
  default:
    return false // 队列满：返回忙信号，交由上游背压处理
  }
}

💡 小贴士

Tell(对应 post)与 TellBlocking(对应 postBlocking)不是两个可以随手挑的 API,而是两种业务语义的显式声明:选 Tell,就是声明「这条消息可以丢、丢了也无伤大雅」(比如心跳、非关键指标上报),对应 credit 耗尽时「丢弃/降级」的策略;选 TellBlocking,就是声明「这条消息一个字都不能丢」(比如本 demo 里的 token),对应 credit 耗尽时「等待」的策略。把这个选择写进代码本身,而不是让它变成一条藏在文档里、容易被后来者忘记的口头约定,这正是 mailbox 把「有界」做成一等公民带来的表达力。

这层同构值得当真,是因为它替你回答了一个极容易被忽略的问题:背压必须端到端贯穿,只在某一层做限流是假背压。Flink 社区花了好几个大版本才把这件事做对——早期只在 TCP 层挡,没有把压力一路传回数据源,结果依然会在中间某个算子的内存里堆出问题。搬到 Actor 世界,这条经验提醒你:如果只在最外层的 HTTP handler 上限流,内部几层 Actor 之间却用无界 channel、或者干脆忽略 Tell 的返回值瞎发消息,压力照样会在中间某个 Actor 的 mailbox 里悄悄堆积,直到那里 OOM——mailbox 有界只是第一步,还得让每一跳都诚实地检查投递是否成功,并把「投递失败」这件事继续向上游传导,这才是「端到端」四个字的真正含义。

Checkpoint ≈ 私有状态即活的检查点

Flink 靠周期性 checkpoint 把算子状态持久化到 state backend,故障后从最近一次成功的 checkpoint 恢复,配合 barrier 对齐机制实现 exactly-once。这一整套机制之所以复杂,根源在于 Flink 的算子状态和算子的执行体是「合一」的——一旦执行体所在的进程或容器没了,状态也跟着没了,所以必须定期往外备份。Actor 的做法在这一点上更轻:只要状态本身活在一个相对稳定的 Actor 身上,而不是活在随时可能崩溃的执行体里,不需要外部备份也能扛住崩溃。

本 demo 的杀手锏就在这里:SessionActor 与 GenWorker 是父子关系,但状态只放在父亲身上——worker(无状态的生成执行体)崩溃、被监督者重启,新实例从零开始,但父 SessionActor 里的 tokens、crashAt 全须全尾地留着,worker 一重启就能从断点续传,外部完全无感。把这句话说重一点:这个 Actor 的私有状态本身就是一份活的检查点——它不是「定期写一份快照供恢复用」,而是从来就没离开过内存,时时刻刻都是最新的,故障恢复不需要「回滚到某个历史时间点」,因为状态压根没有倒退过。这正是「把易崩的执行逻辑与要保住的状态拆到父子两个 Actor」的经典模式,对应 Flink 里「算子状态独立于 task 生命周期,task 重启后从 state backend 恢复」——区别是我们的「state backend」不是外部系统,而就是父 Actor 自己的堆内存。

这套「持续写、随时可读」的状态维护,在源码里落实为每次状态变化后调用一次 persist:每产出一个 token 落一次快照,会话开始、结束时也各落一次——保证任意时刻的内存状态都精确对应最近一次持久化的状态,不存在「两次快照之间丢失若干条消息」的窗口期。

// persist 把当前会话核心状态写入快照。未配置存储时无副作用。
func (s *SessionActor) persist(ctx *actor.Context[SessionMsg]) {
  err := actor.SaveSnapshot(ctx, sessionSnapshot{
    Prompt:  s.prompt,
    Total:   s.total,
    Tokens:  s.tokens,
    Started: s.started,
    Done:    s.done,
    Counter: s.counter,
  })
  if err != nil && err != actor.ErrNoSnapshotStore {
    ctx.Logger().Warn("session snapshot save failed", "session", s.id, "err", err)
  }
}

同一份状态还服务于客户端断线重连重放(对应 Flink 里「从 savepoint 重启作业」),完全不需要外部存储——会话状态本身就是「检查点」。而当状态需要跨进程重启存活(比如整个服务重新部署)时,本笔记的持久化层把这份「活检查点」升格成真正落盘的快照:

// Store[S] 是持久化的【上层接口】：它在字节级 SnapshotStore 之上，提供针对具体状态
// 类型 S 的强类型存取。业务代码面向 Store[S] 编程，存进去、拿出来的都是 S，而不是 []byte
// 或 any——这就是「泛型优先」在持久化层的落地。
//
// 序列化用字节跳动的 sonic（高性能 JSON），而非标准库 encoding/json。
type Store[S any] struct {
  backend SnapshotStore
}

Store[S] 是持久化的上层接口,业务代码面向具体状态类型 S 编程,存进去、拿出来的都是 S 而非字节或 any;底层的 SnapshotStore 则是可插拔的字节级后端(内存、文件,乃至未来的 Redis/RocksDB),对应 Flink 里 state backend 的可插拔性——从纯内存的 state backend 到落盘的 RocksDB state backend,概念上完全对得上。

📝 注意

值得分清两层「检查点」:一层是活检查点——只要 Actor(这里是 SessionActor)本身没有被监督者判定为需要重启,它内存里的状态天然就是最新的,这一层零成本、零配置,对应 Flink 里始终在跑的自动 checkpoint;另一层是落盘快照——需要显式调用 SaveSnapshot、依赖配置好的 SnapshotStore,专门用来扛住「整个进程都没了」这种更重的故障,更接近 Flink 里手动触发、用于版本升级或整体迁移的 savepoint。两者不是非此即彼,而是分层防御:前者管日常的「子 Actor 崩了重启」,后者管小概率的「整个服务重启」。

Restart Strategy ≈ 监督 Directive

Flink 的 RestartStrategy 不是单一策略,而是一族:fixed-delay(固定次数、固定间隔重启)、failure-rate(时间窗口内失败次数超限才彻底放弃)、exponential-delay(重启间隔随失败次数指数拉长,并加抖动防止惊群)——分别应对「瞬时抖动」「持续性故障」「大规模同时失败」这几种不同的失败形态。

本笔记的监督策略几乎是这族策略的直接翻版,而且源码注释里写得明明白白——这不是巧合,是有意为之的对照:RestartStrategy(maxRestarts) 对应 fixed-delay 的思路(最多重启 N 次,超过则永久停止,用来遏制「崩溃—重启—立刻又崩溃」的风暴);BackoffRestartStrategy 按重启次数做指数退避并加抖动,源码注释直接写着「对齐 Flink 的 exponential-delay restart-strategy」;FailureRateStrategy 用滑动时间窗口统计失败次数,超过阈值才放弃,源码注释同样写着「对齐 Flink 的 failure-rate restart-strategy」——比「累计失败次数」更贴近「这个 Actor 是不是真的病了」,因为它会随时间把旧的失败逐渐遗忘掉。

// FailureRateStrategy 返回「时间窗口内失败次数超限则停止」的策略。
//
// 只要 window 时间内的失败次数不超过 maxFailures，就一直重启（可选退避）；
// 一旦窗口内失败太密集，说明是持续性故障而非瞬时抖动，果断停止。
// 对齐 Flink 的 failure-rate restart-strategy——比「累计次数」更贴近真实健康度。
func FailureRateStrategy(maxFailures int, window time.Duration, backoff time.Duration) SupervisorStrategy {
  return StrategyFunc(func(fc FailureContext) Decision {
    cutoff := fc.Now.Add(-window)
    recent := 0
    for _, t := range fc.Failures {
      if t.After(cutoff) {
        recent++
      }
    }
    if recent > maxFailures {
      return Decision{Directive: DirectiveStop}
    }
    return Decision{Directive: DirectiveRestart, Backoff: backoff}
  })
}

这正是这层同构最值得当真的地方:重启策略不是随手选一个数字就能了事的小事,它背后有一整套「失败形态分类学」——Flink 工程师用生产事故换来的分类(瞬时的、持续的、雪崩式的),不需要 Actor 设计者重新踩一遍坑再总结出来,直接照抄这三种策略的形状就够用。如果你的监督策略只有「重启 N 次」这一种武器,遇到短时间内密集失败时就会很尴尬——重启太快,等于在往一个还没恢复的下游继续开火,这正是 FailureRateStrategy 与带抖动的 BackoffRestartStrategy 存在的理由。

Flink「超过重启上限 → 作业 FAILED」正对应监督树里的 DirectiveStop;而 DirectiveEscalate(上报父监督者)则对应 Flink 里 task 失败上升为 region/job 级别的 failover。ManagerActor 在 SpawnChild 时挂上 WithSupervisor(actor.RestartStrategy(5)),就是把这条策略显式钉在了会话这一级——会话最多自愈 5 次,再崩就彻底放弃,而不会拖累整个 ManagerActor,也不会波及其它会话。

小结

• Flink 的算子模型与 Actor 模型在架构层面严丝合缝:keyed state、credit 背压、checkpoint、restart strategy,四组概念一一对应——不是类比,而是同构,而且这层同构的证据不止停留在概念层面,连源码注释都直接写明「对齐 Flink 的某某策略」。
• 这个同构的价值在于可迁移性:流计算框架花十几年趟平的坑——热 key 会拖垮整条流水线、背压必须端到端而非卡在某一层、失败要按形态分类处理——Actor 设计者可以直接抄作业,不必自己再摔一遍跟头。
• 反过来,倘若你的 Actor 系统开始撞上「状态太大放不下内存、要跨节点扩展、要 exactly-once」这类难题,Flink 的解法(增量 checkpoint、RocksDB state backend、两阶段提交)也正是该去取经的方向。
• 下一章我们回到工程本身:动手拆开这个几百行的极简 Actor 运行时,看 RestartStrategy、BackoffRestartStrategy、FailureRateStrategy 这几个刚刚出现过的策略函数究竟是怎么写出来的,把本章说的「keyed state、背压、checkpoint、restart strategy」一一变成可编译、可 -race 检验的代码。