企业官网建设流程全解析

ro 是一个面向 Go 的声明式、可组合的响应式编程库，而Subject正是它实现"热流（Hot Stream）"的核心组件：它既是 Observer（能接收值），又是 Observable（能向多个订阅者广播值）。本文将一次性对比 Publish、Behavior、Replay、Unicast 四种 Subject 的行为差异、内存特性和适用场景，帮你在新手阶段就选对工具。

什么是 Subject：热流的"广播交换机"

在响应式编程中，数据源分为两类：

• 冷流（Cold Observable）：每个订阅者独立触发，各拿一份数据，互不影响；
• 热流（Hot Observable）：数据持续产生并共享，所有订阅者看到的是同一份事件流。

Subject 就是冷流世界通往热流世界的桥梁。它的接口定义在 subject.go 中，同时内嵌了Observable[T]和Observer[T]，并额外提供状态查询能力：

type Subject[T any] interface { Observable[T] // 可被订阅：Subscribe / SubscribeWithContext Observer[T] // 可接收值：Next / Error / Complete HasObserver() bool // 是否还有订阅者 CountObservers() int // 订阅者数量 // ... }

这意味着你可以随时用HasObserver()/CountObservers()检查"现在还有人在听吗"，这在实现事件总线、按需启动数据源等场景非常实用。官方文档的完整介绍见 docs/docs/core/subject.md。

四种 Subject 全对比：一张表看懂区别

先看结论，再逐个拆解：

一句话记忆法：Publish 只播现在，Behavior 送最新值，Replay 发历史，Unicast 一人专线。

PublishSubject：最纯粹的热流广播

PublishSubject是最基础的广播器（定义于 subject_publish.go）：它不缓存任何历史，订阅之前发出的值会直接丢失，订阅者只能收到订阅之后新产生的值。

subject := ro.NewPublishSubject[int]() subject.Next(1) // 此时还没有订阅者，1 被丢弃 subject.Subscribe(ro.OnNext(func(n int) { fmt.Println(n) // 只会收到 2 })) subject.Next(2)

💡提示：ro.NewSubject[T]()就是NewPublishSubject[T]()的别名，日常写哪个都可以（见 subject.go）。

✅适用场景：聊天消息推送、系统事件总线、实时通知等"错过就错过"的即发即弃型事件。它是内存开销最小、性能最好的选择。

BehaviorSubject：新订阅者自动"补发"当前状态

BehaviorSubject在创建时必须提供一个初始值，之后每次Next都会把最新值记下来（实现见 subject_behavior.go）。任何新订阅者接入的瞬间，会立即收到当前最新值，然后继续接收后续更新。

subject := ro.NewBehaviorSubjectstring // 初始值 subject.Subscribe(ro.OnNext(func(s string) { fmt.Println(s) // 立刻打印 "guest" })) subject.Next("alice")

这正是它被称为"状态容器"的原因：任何时刻接入的成员，都自动同步到当前状态，不需要手动查询。

✅适用场景：当前用户身份、主题配置、计数器数值等"任何观察者都需要知道现状"的状态管理场景。

ReplaySubject：回放最近 N 条历史事件

如果说 Behavior 只存"最后 1 条"，ReplaySubject则保存最近 N 条历史（实现见 subject_replay.go），新订阅者会先按时间顺序收到这些历史，再实时接收新值：

subject := ro.NewReplaySubjectstring // 只保留最近 3 条 subject.Next("m1") subject.Next("m2") subject.Next("m3") subject.Subscribe(ro.OnNext(func(s string) { fmt.Println(s) // 依次收到 m1、m2、m3，然后是后续新消息 }))

缓冲区大小bufferSize有两种特殊取值：

• -1（即ReplaySubjectUnlimitedBufferSize）：无限缓冲，保存全部历史；
• 0：不缓存任何值（等效于"只广播、不留史"）。

传入小于 -1 的值会直接 panic，错误信息为ro.NewReplaySubject: bufferSize must be greater or equal to -1（见 errors.go）。

⚠️注意：缓冲满了之后，最老的值会被循环覆盖丢弃。

✅适用场景：聊天室历史消息、股票行情快照、通知列表——新加入者需要一点"上下文"才能看懂后续事件的场景。

UnicastSubject：一对一的"专线管道"

UnicastSubject是最特别的一种（实现见 subject_unicast.go）：它只接受一个订阅者，语义是"把事件攒起来，等唯一的那个消费者上线后先回放、再切换为实时转发"。

subject := ro.NewUnicastSubjectint // 最多排队 10 个事件 subject.Next(1) // 尚无订阅者，事件进入排队 subject.Next(2) subject.Subscribe(ro.OnNext(func(n int) { fmt.Println(n) // 先收到排队的 1、2，之后收到实时值 }))

它的三个关键行为值得牢记：

1. 先排队、后回放：订阅发生前的事件缓存在内部，订阅瞬间回放给这唯一的订阅者，随后缓存被清空，不再重复投递；
2. 拒绝并发订阅：如果第二个订阅者在第一位还在线时接入，会收到错误ro.UnicastSubject: a single subscriber accepted（ErrUnicastSubjectConcurrent，见 errors.go）；
3. 缓冲可限量：同样支持-1（UnicastSubjectUnlimitedBufferSize）表示无限排队，超限时最老的事件被丢弃。

✅适用场景：生产者先跑、消费者稍后才就绪的"接力"场景，比如上游数据生成器与下游唯一处理模块之间的缓冲桥梁——用 Unicast 既避免丢早期数据，又不会像 Replay 那样让每个新来者都重放全量历史。

顺手认识：AsyncSubject 只发"最终结果"

除了标题中的四兄弟，ro 还提供AsyncSubject（subject_async.go）：它只保存最后一个值，并且只有在Complete()之后才发射，因此每个订阅者最多收到 0 或 1 个值。适合"一次异步计算的最终产物"这类语义，例如把多个竞争请求收敛为一个结果。

避坑指南：三个最容易踩的坑

📌坑一：无限 Replay 内存泄漏。ro.NewReplaySubjectT会永久保存全部历史，长期运行的服务中慎用，务必按业务上限设置 N。

📌坑二：Publish 的"早到"数据会丢。如果你希望订阅者接入时拿到"最新状态"，却用了 PublishSubject，会收到空——这时应换成 BehaviorSubject。

📌坑三：Unicast 被多人订阅。它是单订阅者设计，多人接入会直接报错；需要一对多广播时请选 Publish / Replay。

另外，所有 Subject 在Error()或Complete()之后就进入终止态：后续Next的值会被丢弃，新订阅者只会收到缓存的终态（错误或直接完成）。可以用IsClosed()、HasThrown()、IsCompleted()做防御性检查。

总结：一分钟选型速查

ro 的 Subject 家族把响应式编程中最核心的"多播 + 缓冲"策略做成了四五个开箱即用的类型，配合 docs/docs/core/subject.md 中的完整示例，足以支撑聊天室、事件总线、状态共享、行情推送等常见事件驱动应用。理解每种 Subject"回放什么、保留多少、服务几人"这三个问题，你就掌握了热流选型的全部要点。