golang-samber-ro

角色定位： 你是一名 Go 工程师，在数据以异步或无限流动时会选择响应式流。你使用 samber/ro 构建声明式管道，而不是手动连接 goroutine/channel，但也知道何时简单的切片 + samber/lo 就足够了。

思考模式： 设计高级响应式管道，或在冷/热 observable、subject 和组合运算符之间做选择时，要尽可能全面地推理——架构选择错误会导致资源泄漏或事件遗漏。在 Claude Code 中，使用 ultrathink 显式触发扩展思考。

samber/ro — Go 响应式流

ReactiveX 的 Go 实现。以泛型为先、类型安全且可组合的异步数据流管道，支持自动背压、错误传播、上下文集成和资源清理。包含 150 多个运算符、5 种 subject 和 40 多个插件。

官方资源：

• github.com/samber/ro
• ro.samber.dev
• pkg.go.dev/github.com/samber/ro

本技能并未涵盖所有内容——更多信息请参阅库文档和代码示例：

• 如需 Go 包文档、符号、版本、导入方和已知漏洞，→ 请参阅 samber/cc-skills-golang@golang-pkg-go-dev 技能（godig），查询 Go 包信息时优先于 Context7。
• 如需在自己的代码中查找此库的用法（定义、调用位置、诊断信息），→ 请参阅 samber/cc-skills-golang@golang-gopls 技能（gopls）。
• 对于未在 pkg.go.dev 上索引的文档，Context7 仍可作为备用方案。

为什么选择 samber/ro（流与切片）

对于复杂的异步管道，Go channel + goroutine 会变得难以管理：需要手动关闭 channel，goroutine 生命周期管理冗长，在嵌套 select 之间传播错误也很繁琐，而且没有可组合的运算符。samber/ro 通过声明式、可链式调用的流运算符解决了这些问题。

适用场景：

场景	工具	原因
转换切片（map、filter、reduce）	samber/lo	有限、同步、急切求值——无需流带来的额外开销
简单的 goroutine 扇出并处理错误	errgroup	标准库、轻量，对于有界并发已足够
无限事件流（WebSocket、定时器、文件监视器）	samber/ro	声明式管道，支持背压、重试、超时和组合
从多个异步源实时丰富数据	samber/ro	CombineLatest/Zip 可组合相关流，无需手动使用 select
多个消费者共享一个源的发布/订阅	samber/ro	热 observable（Share/Subjects）可原生处理多播

关键区别：lo 与 ro

方面	samber/lo	samber/ro
数据	有限切片	无限流
执行	同步、阻塞	异步、非阻塞
求值	急切（分配中间切片）	惰性（数据到达时逐项处理）
时序	即时	感知时间（delay、throttle、interval、timeout）
错误模型	每次调用返回 (T, error)	错误通过管道中的错误通道传播
使用场景	集合转换	事件驱动、实时、异步管道

安装

go get github.com/samber/ro

核心概念

四个构建模块：

1. Observable — 随时间发出值的数据源。默认是冷 observable：每个订阅者都会触发一次独立的全新执行
2. Observer — 包含三个回调的消费者：onNext(T)、onError(error)、onComplete()
3. Operator — 将一个 observable 转换为另一个 observable 的函数，通过 Pipe 链式调用
4. Subscription — observable 与 observer 之间的连接。调用 .Wait() 阻塞，或调用 .Unsubscribe() 取消订阅

observable := ro.Pipe2(
    ro.RangeWithInterval(0, 5, 1*time.Second),
    ro.Filter(func(x int) bool { return x%2 == 0 }),
    ro.Map(func(x int) string { return fmt.Sprintf("even-%d", x) }),
)

observable.Subscribe(ro.NewObserver(
    func(s string) { fmt.Println(s) },      // onNext
    func(err error) { log.Println(err) },    // onError
    func() { fmt.Println("Done!") },         // onComplete
))
// Output: "even-0", "even-2", "even-4", "Done!"

// Or collect synchronously:
values, err := ro.Collect(observable)

冷与热可观测对象

冷（默认）：每次调用 .Subscribe() 都会启动一个新的独立执行。安全且可预测——默认使用。

热：多个订阅者共享同一次执行。适用于数据源开销较大的情况（WebSocket、数据库轮询），或订阅者必须看到相同事件的情况。

转换方式	行为
Share()	将冷可观测对象转换为热可观测对象，并采用引用计数。最后一个订阅者取消订阅时会停止执行
ShareReplay(n)	与 Share 相同，此外还会缓存最近的 N 个值，供较晚订阅的订阅者接收
Connectable()	将冷可观测对象转换为热可观测对象，但会等待显式调用 .Connect()
Subjects	原生为热；可直接调用 .Send()、.Error()、.Complete()

Subject	构造函数	重放行为
PublishSubject	NewPublishSubject[T]()	无——较晚订阅的订阅者会错过过往事件
BehaviorSubject	NewBehaviorSubject[T](initial)	向新订阅者重放最近的值
ReplaySubject	NewReplaySubject[T](bufferSize)	重放最近的 N 个值
AsyncSubject	NewAsyncSubject[T]()	仅在完成时发出最后一个值
UnicastSubject	NewUnicastSubject[T](bufferSize)	仅允许单个订阅者

有关 subject 详情和热可观测对象模式，请参阅 Subjects Guide。

运算符速查

类别	主要运算符	用途
创建	Just、FromSlice、FromChannel、Range、Interval、Defer、Future	从各种数据源创建可观测对象
转换	Map、MapErr、FlatMap、Scan、Reduce、GroupBy	转换或累积流中的值
筛选	Filter、Take、TakeLast、Skip、Distinct、Find、First、Last	有选择地发出值
组合	Merge、Concat、Zip2–Zip6、CombineLatest2–CombineLatest5、Race	合并多个可观测对象
错误	Catch、OnErrorReturn、OnErrorResumeNextWith、Retry、RetryWithConfig	从错误中恢复
计时	Delay、DelayEach、Timeout、ThrottleTime、SampleTime、BufferWithTime	控制发出值的时机
副作用	Tap/Do、TapOnNext、TapOnError、TapOnComplete	在不更改流的情况下进行观察
终结	Collect、ToSlice、ToChannel、ToMap	将流收集为 Go 类型

使用类型化的 Pipe2、Pipe3……Pipe25，可在运算符链中实现编译时类型安全。无类型的 Pipe 使用 any，会失去类型检查。

完整的运算符目录（含 150 多个运算符及其签名），请参阅 Operators Guide。

常见错误

错误	失败原因	修正方法
使用 ro.OnNext()，但未提供错误处理程序	错误会被静默丢弃——缺陷会隐藏在生产环境中	使用 ro.NewObserver(onNext, onError, onComplete)，并提供全部 3 个回调
使用无类型的 Pipe()，而非 Pipe2/Pipe3	失去编译时类型安全，错误会在运行时出现	在类型化的运算符链中使用 Pipe2、Pipe3……Pipe25
忘记在无限流上调用 .Unsubscribe()	Goroutine 泄漏——可观测对象会永远运行	使用 TakeUntil(signal)、取消 context，或显式调用 Unsubscribe()
在冷可观测对象已足够时仍使用 Share()	增加不必要的复杂性，也更难推断生命周期	仅在多个使用者需要相同流时才使用热可观测对象
使用 samber/ro 转换有限切片	对同步操作而言，流会产生额外开销（Goroutine、订阅）	使用 samber/lo——它更简单、更快，且专为切片设计
未传递 context 以支持取消	流会忽略关闭信号，导致终止时资源泄漏	在管道中串联 ContextWithTimeout 或 ThrowOnContextCancel

最佳实践

1. 始终处理全部三种事件——使用 NewObserver(onNext, onError, onComplete)，而不仅仅是 OnNext。未处理的错误会导致数据悄然丢失
2. 同步消费时使用 Collect()——当流是有限的且你需要 []T 时，Collect 会阻塞直到流结束，并返回切片和错误
3. 优先使用类型化的 Pipe 函数——Pipe2、Pipe3...Pipe25 会在编译时捕获类型不匹配。仅在动态运算符链中使用非类型化的 Pipe
4. 为无限流设置边界——使用 Take(n)、TakeUntil(signal)、Timeout(d) 或取消 context。无界流会导致 goroutine 泄漏
5. 使用 Tap/Do 实现可观测性——记录日志、跟踪或统计数据流发出的值，而不改变数据流。串联 TapOnError 以监控错误
6. 简单转换优先使用 samber/lo——如果数据是有限切片，且你需要 Map/Filter/Reduce，请使用 lo。当数据随时间到达、来自多个来源，或需要重试/超时时，再使用 ro

插件生态系统

40 多个插件为 ro 扩展了特定领域的运算符：

类别	插件	导入路径前缀
编码	JSON、CSV、Base64、Gob	plugins/encoding/...
网络	HTTP、I/O、FSNotify	plugins/http、plugins/io、plugins/fsnotify
调度	Cron、ICS	plugins/cron、plugins/ics
可观测性	Zap、Slog、Zerolog、Logrus、Sentry、Oops	plugins/observability/...、plugins/samber/oops
速率限制	Native、Ulule	plugins/ratelimit/...
数据	Bytes、Strings、Sort、Strconv、Regexp、Template	plugins/bytes、plugins/strings 等
系统	Process、Signal	plugins/proc、plugins/signal

如需查看包含导入路径和用法示例的完整插件目录，请参阅 插件生态系统。

如需了解真实场景中的响应式模式（重试+超时、WebSocket 扇出、优雅关闭、流组合），请参阅 模式。

如果你在 samber/ro 中遇到错误或意外行为，请在 github.com/samber/ro/issues 提交问题。

交叉引用

• → 有限切片转换（Map、Filter、Reduce、GroupBy）请参阅 samber/cc-skills-golang@golang-samber-lo 技能——数据已在切片中时使用 lo
• → 可与 ro 管道组合的单子类型（Option、Result、Either）请参阅 samber/cc-skills-golang@golang-samber-mo 技能
• → 内存缓存（也可作为 ro 插件使用）请参阅 samber/cc-skills-golang@golang-samber-hot 技能
• → 当响应式流显得过于复杂时，goroutine/channel 模式请参阅 samber/cc-skills-golang@golang-concurrency 技能
• → 在生产环境中监控响应式管道请参阅 samber/cc-skills-golang@golang-observability 技能