golang-samber-ro

角色設定： 你是一位 Go 工程師，當資料以非同步或無限方式流動時，你會優先使用反應式串流。你使用 samber/ro 來建立宣告式管線，取代手動的 goroutine/channel 接線，但你也知道何時簡單的 slice + samber/lo 就足夠了。

思考模式： 在設計進階反應式管線或選擇冷/熱 Observable、Subject 以及組合運算子時，使用 ultrathink。錯誤的架構會導致資源洩漏或錯失事件。

samber/ro — Go 的反應式串流

Go 實作的 ReactiveX。以泛型為優先、型別安全、可組合的管線，用於非同步資料串流，具備自動背壓、錯誤傳播、context 整合以及資源清理。150+ 運算子、5 種 Subject 類型、40+ 外掛。

官方資源：

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

此技能並非詳盡無遺。請參閱函式庫文件與程式碼範例以取得更多資訊。Context7 可作為探索平台。如需 Go 套件文件、版本、符號及已知漏洞，→ 請參閱 samber/cc-skills-golang@golang-pkg-go-dev 技能。

為什麼選擇 samber/ro（串流 vs Slice）

對於複雜的非同步管線，Go channel + goroutine 會變得難以處理：手動關閉 channel、冗長的 goroutine 生命週期、跨巢狀 select 的錯誤傳播，以及缺乏可組合的運算子。samber/ro 透過宣告式、可鏈結的串流運算子解決了這個問題。

何時使用哪種工具：

情境	工具	原因
轉換 slice（map、filter、reduce）	samber/lo	有限、同步、立即執行 — 不需要串流開銷
簡單的 goroutine 扇出並處理錯誤	errgroup	標準函式庫、輕量、足以應付有限並發
無限事件串流（WebSocket、ticker、檔案監控）	samber/ro	宣告式管線，具備背壓、重試、超時、組合
來自多個非同步來源的即時資料擴充	samber/ro	CombineLatest/Zip 組合相依串流，無需手動 select
多個消費者共享一個來源的發布/訂閱	samber/ro	熱 Observable（Share/Subjects）原生支援多播

lo 與 ro 的主要差異

面向	samber/lo	samber/ro
資料	有限 slice	無限串流
執行	同步、阻塞	非同步、非阻塞
評估	立即（分配中間 slice）	惰性（項目到達時處理）
時序	立即	時間感知（delay、throttle、interval、timeout）
錯誤模型	每次呼叫回傳 (T, error)	錯誤透過管線傳播
使用案例	集合轉換	事件驅動、即時、非同步管線

安裝

go get github.com/samber/ro

核心概念

四個建構區塊：

1. Observable — 隨時間發出值的資料來源。預設為冷：每次 .Subscribe() 都會觸發獨立的從頭執行
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
))
// 輸出："even-0"、"even-2"、"even-4"、"Done!"

// 或者同步收集：
values, err := ro.Collect(observable)

冷 vs 熱 Observable

冷（預設）：每次 .Subscribe() 都會啟動一個新的獨立執行。安全且可預測 — 預設使用。

熱：多個訂閱者共享單一執行。當來源昂貴（WebSocket、DB 輪詢）或訂閱者必須看到相同事件時使用。

轉換方式	行為
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 詳細資訊和熱 Observable 模式，請參閱 Subjects 指南。

運算子快速參考

類別	主要運算子	用途
建立	Just、FromSlice、FromChannel、Range、Interval、Defer、Future	從各種來源建立 observable
轉換	Map、MapErr、FlatMap、Scan、Reduce、GroupBy	轉換或累積串流值
過濾	Filter、Take、TakeLast、Skip、Distinct、Find、First、Last	選擇性地發出值
組合	Merge、Concat、Zip2–Zip6、CombineLatest2–CombineLatest5、Race	合併多個 observable
錯誤	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+ 運算子及其簽名），請參閱 運算子指南。

常見錯誤

錯誤	失敗原因	修正
使用 ro.OnNext() 但沒有錯誤處理器	錯誤被靜默丟棄 — 錯誤隱藏在生產環境中	使用 ro.NewObserver(onNext, onError, onComplete) 並提供所有 3 個回呼
使用未型別化的 Pipe() 而非 Pipe2/Pipe3	失去編譯時期型別安全，錯誤在執行時期浮現	使用 Pipe2、Pipe3...Pipe25 進行型別化的運算子鏈
在無限串流上忘記 .Unsubscribe()	Goroutine 洩漏 — observable 永遠執行	使用 TakeUntil(signal)、context 取消或明確的 Unsubscribe()
在冷 Observable 足夠時使用 Share()	不必要的複雜性，更難推理生命週期	僅在多個消費者需要相同串流時使用熱 Observable
對有限 slice 轉換使用 samber/ro	同步操作卻有串流開銷（goroutine、訂閱）	使用 samber/lo — 更簡單、更快，且專為 slice 設計
未傳播 context 以進行取消	串流忽略關閉訊號，導致終止時資源洩漏	在管線中鏈結 ContextWithTimeout 或 ThrowOnContextCancel

最佳實踐

1. 始終處理所有三個事件 — 使用 NewObserver(onNext, onError, onComplete)，而不僅僅是 OnNext。未處理的錯誤會導致靜默資料遺失
2. 對同步消費使用 Collect() — 當串流有限且你需要 []T 時，Collect 會阻塞直到完成並回傳 slice 和錯誤
3. 偏好型別化的 Pipe 函式 — Pipe2、Pipe3...Pipe25 在編譯時期捕獲型別不匹配。保留未型別化的 Pipe 用於動態運算子鏈
4. 限制無限串流 — 使用 Take(n)、TakeUntil(signal)、Timeout(d) 或 context 取消。無限制的串流會洩漏 goroutine
5. 使用 Tap/Do 進行可觀測性 — 記錄、追蹤或測量發送而不改變串流。鏈結 TapOnError 以進行錯誤監控
6. 對簡單轉換偏好 samber/lo — 如果資料是有限 slice 且你需要 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 開啟 issue。

交叉參考

• → 請參閱 samber/cc-skills-golang@golang-samber-lo 技能以了解有限 slice 轉換（Map、Filter、Reduce、GroupBy）— 當資料已在 slice 中時使用 lo
• → 請參閱 samber/cc-skills-golang@golang-samber-mo 技能以了解可與 ro 管線組合的單子型別（Option、Result、Either）
• → 請參閱 samber/cc-skills-golang@golang-samber-hot 技能以了解記憶體快取（也可作為 ro 外掛使用）
• → 請參閱 samber/cc-skills-golang@golang-concurrency 技能以了解當反應式串流過度設計時的 goroutine/channel 模式
• → 請參閱 samber/cc-skills-golang@golang-observability 技能以了解在生產環境中監控反應式管線