RxFeedback 源码深度解读

RxSwift 生态中最精妙的架构 — 从反馈循环到单向数据流

1. 项目概述

RxFeedback 是由 RxSwift 创始人 Krunoslav Zaher（GitHub @kzaher）专为 RxSwift 设计的最简洁架构。整个库的核心源码仅约 300 行，却实现了一套完整的单向数据流系统。
代码地址：https://github.com/NoTests/RxFeedback.swift

为什么需要 RxFeedback？

在 RxSwift 项目中，开发者常常面临一个问题：如何组织复杂的响应式代码？ 直接使用 RxSwift 可以写出任意复杂的流，但缺乏结构约束会导致：

副作用与业务逻辑混在一起，难以测试
状态变更分散在多个地方，难以追踪
循环依赖难以建模和控制

RxFeedback 的答案：用反馈循环（Feedback Loop）将一切纳入声明式系统。

技术概况

项目	说明
作者	Krunoslav Zaher（RxSwift 创始人）
许可证	MIT
平台	iOS / macOS / tvOS / watchOS
依赖	RxSwift, RxCocoa
Swift 版本	5.0+
安装方式	CocoaPods / Carthage / SPM

2. 核心设计理念

2.1 三个核心概念

RxFeedback 将一切系统建模为三个元素的组合：

┌─────────────────────────────────────────────────────────┐
│                     RxFeedback 三要素                     │
├────────────┬────────────────┬─────────────────────────────┤
│   State    │     Event      │         Feedback            │
│   状态     │     事件       │         副作用              │
├────────────┼────────────────┼─────────────────────────────┤
│ 系统的完整  │ 已发生的事实    │ 观察状态 → 产生事件          │
│ 快照，通常  │ 驱动状态变化    │ (Observable<State>          │
│ 是值类型    │ 的不可变数据    │  → Observable<Event>)       │
└────────────┴────────────────┴─────────────────────────────┘

2.2 反馈循环的本质

反馈循环是控制论中的经典概念。在 RxFeedback 中：

State 流向 Feedback，Feedback 产生 Event，Event 通过 reduce 更新 State，形成一个闭环。

2.3 设计哲学

原则	说明
声明式	系统行为在编译期完整声明，`subscribe` 后自动运行
纯函数驱动	`reduce` 是纯函数 `(State, Event) → State`，易于测试
副作用完全分离	网络请求、数据库等副作用被隔离在 `feedback` 闭包中
单向数据流	`UI → Event → State → UI`，数据流向唯一且可预测
可组合	多个 feedback 可独立观察状态并产生事件
DI 友好	纯函数和反馈定义天然适合依赖注入

3. 架构全景图

┌──────────────────────────────────────────────────────────────────────┐
│                        RxFeedback System                              │
│                                                                       │
│  ┌─────────────┐     ┌─────────────┐     ┌─────────────────────────┐ │
│  │     UI      │────→│   Events    │────→│    reduce(State, Event) │ │
│  │ (按钮/输入)  │     │ (用户行为)   │     │     → new State          │ │
│  └─────────────┘     └─────────────┘     └───────────┬─────────────┘ │
│        ↑                                              │               │
│        │                                              ↓               │
│  ┌─────┴──────────┐                        ┌─────────────────────┐   │
│  │   UI Binding   │                        │  feedback1(State)   │   │
│  │ (label.text =) │                        │  → Observable<Event> │   │
│  └────────────────┘                        └──────────┬──────────┘   │
│        ↑                                              │               │
│        │                               ┌──────────────┼──────────┐   │
│        │                               │              ↓          │   │
│        │                        ┌──────┴──────┐ ┌─────┴──────┐   │   │
│        │                        │  Network    │ │  Timer     │   │   │
│        │                        │  Request    │ │  Side      │   │   │
│        │                        │  Effect     │ │  Effect    │   │   │
│        │                        └─────────────┘ └────────────┘   │   │
│        │                               │              │          │   │
│        │                               └──────┬───────┘          │   │
│        │                                      ↓                  │   │
│        └────────────────────────── Observable<State> ←───────────┘   │
│                              (Driver / Observable)                    │
└──────────────────────────────────────────────────────────────────────┘

4. 源码入口 — 核心类型定义

4.1 `Feedback` 类型别名

整个库的基石，一行定义就揭示了全部设计意图：

/// 反馈循环：接收当前状态流，产生新的事件流
public typealias Feedback<State, Event> = (Observable<State>) -> Observable<Event>

解读：

输入：Observable<State> — 当前系统的状态序列。每次状态更新都会推送给 feedback。
输出：Observable<Event> — 由该 feedback 产生的事件流。例如网络请求的响应、定时器触发、数据库查询结果。
本质：一个 高阶 Observable 函数，它可以订阅状态来决定何时、如何产生副作用事件。

4.2 `Query` 类型别名（内部）

feedback 内部常用的模式被抽象为 Query：

/// 查询：当状态满足条件时，执行请求并映射为事件
public typealias Query<State, Event> = 
    (Driver<State>) -> Driver<Event>?

解读：

查询是反馈的一种特化形式 — 仅在特定状态条件下触发。
返回 nil 时表示"当前不做任何事"，返回一个 observable 时表示"执行这个副作用并映射为事件"。
典型场景：if 状态要求加载数据 → 发起网络请求 → 返回成功/失败事件。

4.3 `Bindings` 类型别名

为了在 UI 层绑定，定义了两种绑定形态：

/// 订阅绑定：订阅一个 Observable 并执行副作用（通常是 UI 更新）
public typealias Subscribe<State> = (Observable<State>) -> Disposable

/// 观察绑定：将 UI 事件源注册为系统的 Event
public typealias Mutate<Event> = (Observable<Event>) -> Disposable

解读：

Subscribe<State> 负责输出 — 将状态变化反映到 UI（如设置 label.text）。
Mutate<Event> 负责输入 — 将 UI 事件注册进系统（如按钮点击 → Event）。
两者返回 Disposable，便于生命周期管理。

5. 系统操作符 — `system` 函数

这是 RxFeedback 的心脏，整个库只有这一个核心 operator。

5.1 函数签名

public static func system<State, Event>(
    initialState: State,
    reduce: @escaping (State, Event) -> State,
    feedback: Feedback<State, Event>...
) -> Observable<State>

5.2 源码实现解析

public static func system<State, Event>(
    initialState: State,
    reduce: @escaping (State, Event) -> State,
    feedback: Feedback<State, Event>...
) -> Observable<State> {
    
    return Observable<State>.deferred {
        // ──── 第一步：创建状态重放 Subject ────
        // 为什么用 ReplaySubject(bufferSize: 1)？
        // → 每个 feedback 订阅时都能获取到最新状态
        // → 新订阅者不会错过当前状态
        let replaySubject = ReplaySubject<State>.create(bufferSize: 1)
        
        // ──── 第二步：构建事件源 ────
        // 所有 feedback 产生的事件合并为一个流
        let events: Observable<Event> = Observable<Event>
            .merge(feedback.map { $0(replaySubject.asObservable()) })
            //             ↑ 将状态流注入每个 feedback
            //             ↑ 每个 feedback 返回自己的事件流
            //             ↑ merge 将它们全部合并
            
            // 可选：额外的事件源
            // .merge(scheduledEvents)
            // 用于注入外部事件（如定时触发、系统通知）
        
        // ──── 第三步：构建状态流 ────
        return events
            // scan 是核心：用 reduce 将 Event 序列折叠为 State 序列
            // scan(initialState) { state, event in reduce(state, event) }
            .scan(initialState, accumulator: reduce)
            // 确保初始状态立即发出
            .startWith(initialState)
            // ──── 第四步：建立反馈循环 ────
            // 每次状态更新 → 推送给 replaySubject 
            // → feedback 收到新状态 → 产生新事件 → scan 更新状态 → 循环
            .do(onNext: { state in
                replaySubject.onNext(state)
            })
    }
}

5.3 逐行剖析

第 1 层：`Observable.deferred`

return Observable<State>.deferred {

为什么用 deferred？

deferred 确保每次新订阅都创建一个全新的反馈循环系统。如果不使用 deferred，多个订阅者会共享同一个 replaySubject，导致状态混乱。使用 deferred 后每个订阅者都有自己独立的系统实例。

第 2 层：`ReplaySubject` 状态中继

let replaySubject = ReplaySubject<State>.create(bufferSize: 1)

参数	含义
`ReplaySubject`	可以重放历史值的 Subject，新订阅者收到最近的值
`bufferSize: 1`	只保留最新的 1 个状态值
作用	作为"状态总线"，将 scan 输出的状态分发给所有 feedback

为什么是 bufferSize: 1？

feedback 只需要当前状态来判断下一步动作，历史状态不需要。如果保留更多，可能导致内存泄漏（状态对象一直无法释放）。

第 3 层：事件合并

let events: Observable<Event> = Observable<Event>
    .merge(feedback.map { $0(replaySubject.asObservable()) })

feedback1: (Observable<State>) → Observable<Event>  ──┐
feedback2: (Observable<State>) → Observable<Event>  ──┼── merge ──→ 统一事件流
feedback3: (Observable<State>) → Observable<Event>  ──┘

关键技巧：replaySubject.asObservable() 将 Subject 转换为只读的 Observable，防止 feedback 意外修改状态。每个 feedback 拿到的是同一个状态流的只读视图。

第 4 层：`scan` — 状态折叠

.scan(initialState, accumulator: reduce)

// scan 的行为等价于：
// State[0] = initialState
// State[1] = reduce(State[0], Event[0])
// State[2] = reduce(State[1], Event[1])
// ...

初始状态 → Event0 → State0 → Event1 → State1 → Event2 → State2 → ...

scan 是 RxFeedback 中最关键的操作符。它将事件序列折叠为状态序列，保证了：

每个状态都是从前一个状态 + 最新事件确定性地计算得出
reduce 是纯函数，相同输入必然产生相同输出
状态变更是原子的 — 一次 reduce 调用完成所有字段更新

第 5 层：`startWith` — 初始状态立即发出

.startWith(initialState)

确保订阅者立即收到初始状态，即使还没有任何事件发生。这对于 UI 初始化至关重要 — 页面应该在显示时就渲染初始状态。

第 6 层：`do(onNext:)` — 构建反馈回路

.do(onNext: { state in
    replaySubject.onNext(state)
})

这是整个架构的核心闭环：

scan 产出 State
    │
    ↓  do(onNext:)
    │
replaySubject.onNext(state)
    │
    ↓  分发给所有 feedback
    │
feedback₁(state$) → event₁$
feedback₂(state$) → event₂$
    │
    ↓  merge
    │
events → scan → State → 循环...

为什么用 do 而不是显式订阅？

do 操作符是副作用插入点，不改变流的行为，只在数据经过时执行动作。用于建立反馈循环非常合适：状态流照常向下传递，但"顺便"通知 feedback。这也保证了同一个状态会被 UI 和 feedback 同时收到。

5.4 补充：扩展版 `system` — 支持 `scheduledObservable`

RxFeedback 还提供了支持外部事件注入的版本：

public static func system<State, Event>(
    initialState: State,
    reduce: @escaping (State, Event) -> State,
    feedback: Feedback<State, Event>...,
    // 由外部调度器驱动的定时事件
    scheduledObservable: Observable<Event>
) -> Observable<State>

唯一区别在于事件合并阶段：

let events = Observable<Event>.merge(
    feedback.map { $0(replaySubject.asObservable()) } + [scheduledObservable]
)

这使得外部定时事件（如心跳、轮询）能注入系统，而不需要将它们包装成一个 feedback。

6. 辅助方法 — `react` 与 `bind`

6.1 `react` — 声明式副作用触发

public static func react<State, Event>(
    request: @escaping (State) -> Bool,
    effects: @escaping (State) -> Observable<Event>
) -> Feedback<State, Event> {
    
    return { state$ in
        return state$
            // 仅当 request 返回 true 时才触发
            .filter(request)
            // 切换到一个专门的调度器执行副作用
            .observeOn(/* 后台调度器 */)
            // 执行副作用并映射为事件
            .flatMapLatest { state in
                effects(state)
            }
    }
}

解读：

参数	作用
`request: (State) -> Bool`	判断是否需要执行副作用
`effects: (State) -> Observable<Event>`	需要执行的副作用逻辑
返回	一个新的 `Feedback<State, Event>`

典型用法 — 分页加载：

react(
    request: { $0.shouldLoadNextPage },  // 条件
    effects: { state in                    // 副作用
        api.search(state.query, page: state.nextPage)
            .map { Event.response($0) }     // 映射为 Event
    }
)

关键设计 — flatMapLatest：

使用 flatMapLatest 而非 flatMap，确保新请求到来时自动取消旧请求。例如用户快速切换搜索词，旧的搜索请求会被自动丢弃。

6.2 `bind` — UI 绑定辅助方法

public static func bind<State, Event>(
    _ bindings: @escaping (ObservableSchedulerContext<State>) -> ObservableSchedulerContext<Event>
) -> Feedback<State, Event> {
    
    return { state$ in
        Observable<Event>.using({
            let scope = ScopedSubject()
            let context = ObservableSchedulerContext<State>(
                source: state$,
                scheduler: MainScheduler.asyncInstance
            )
            let events = bindings(context)
            // 将 context 中的绑定转换为 Event 流
            return ...
        }, observableFactory: { scope in
            // 返回绑定产生的事件
            return scope.asObservable()
        })
    }
}

解读：

bind 的独特之处在于它使用了 ObservableSchedulerContext（下文详解），这是 RxFeedback 与 RxCocoa 集成的桥梁。它允许在 feedback 内部使用 Driver 语义的绑定。

7. 与 RxCocoa 集成 — `ObservableSchedulerContext`

7.1 为什么需要 Context？

普通 Observable<State> 在 RxSwift 中没有调度器信息。在 UI 开发中，绑定必须发生在主线程。ObservableSchedulerContext 解决了这个问题：

public struct ObservableSchedulerContext<Element> {
    let source: Observable<Element>
    let scheduler: ImmediateSchedulerType
}

7.2 `bind` 方法

extension ObservableSchedulerContext {
    /// 将状态变化绑定到一个 observer（UI 更新）
    func bind<Observer: ObserverType>(
        to observer: Observer
    ) -> Disposable where Observer.Element == Element {
        return source
            .observeOn(scheduler)   // 切换到指定调度器
            .subscribe(observer)     // 订阅观察者
    }
}

7.3 `bind` + `map` 组合

func bind<Result>(
    to binder: @escaping (Observable<Result>) -> Disposable,
    _ transform: @escaping (Element) -> Result
) -> Disposable {
    return source
        .observeOn(scheduler)
        .map(transform)
        .subscribe(onNext: { value in
            _ = binder(Observable.just(value))
        })
}

7.4 实际使用示例

RxFeedback.system(
    initialState: State.empty,
    reduce: State.reduce,
    feedback: bind { state in
        // 从 state 中提取 UI 绑定
        let subscriptions = [
            // 状态变化 → UI 更新
            state.map { $0.query }.bind(to: textField.rx.text),
            state.map { $0.results }.bind(to: tableView.rx.items),
        ]
        // UI 事件 → 系统 Event
        let events = [
            textField.rx.text.orEmpty
                .map(Event.searchChanged),
        ]
        return Bindings(subscriptions: subscriptions, events: events)
    }
)

8. 完整工作流程剖析

8.1 时间线视图

8.2 内存与生命周期

当外部取消对 system() 返回的 Observable 的订阅时，整个系统（包括所有 feedback 中的副作用）都会被级联释放。这得益于 RxSwift 的 Disposable 机制。

9. 示例解读 — GitHub 搜索

这是 RxFeedback 仓库中的经典示例，完整展示了架构的实际应用。

9.1 状态定义

struct State {
    var searchText: String        // 当前搜索词
    var results: [Repository]     // 搜索结果
    var nextPage: Int?            // 下一页页码（nil 表示没有更多）
    var isLoading: Bool           // 是否正在加载
    var error: Error?             // 错误信息
}

9.2 事件定义

enum Event {
    case searchChanged(String)            // 搜索词改变
    case response([Repository], Int?)     // 收到响应 + 下一页
    case loadingFailed(Error)             // 加载失败
}

9.3 Reducer — 纯函数状态更新

extension State {
    static func reduce(state: State, event: Event) -> State {
        var newState = state
        
        switch event {
        case .searchChanged(let text):
            newState.searchText = text
            newState.results = []
            newState.nextPage = 1          // 重置分页
            newState.isLoading = true       // 开始加载
            
        case .response(let repos, let nextPage):
            newState.results += repos       // 追加结果
            newState.nextPage = nextPage
            newState.isLoading = false
            
        case .loadingFailed(let error):
            newState.error = error
            newState.isLoading = false
        }
        
        return newState
    }
}

9.4 Feedback 定义

// 反馈1：当需要加载时，发起网络请求
let loadPage: Feedback<State, Event> = react(
    request: { state in
        // 条件：搜索词非空 且 正在加载
        !state.searchText.isEmpty && state.isLoading
    },
    effects: { state -> Observable<Event> in
        return GitHubSearch.search(
            query: state.searchText,
            page: state.nextPage ?? 1
        )
        .map { Event.response($0.repos, $0.nextPage) }
        .asObservable()
        .catchError { .just(Event.loadingFailed($0)) }
    }
)

9.5 UI 绑定

let uiFeedback: Feedback<State, Event> = bind(self) { me, state in
    let subscriptions = [
        // 状态 → UI
        state.map { $0.results }.bind(to: me.tableView.rx.items(...)),
        state.map { $0.isLoading }.bind(to: me.activityIndicator.rx.isAnimating),
        state.map { $0.searchText }.bind(to: me.searchTextField.rx.text),
    ]
    let events = [
        // UI → 事件
        me.searchTextField.rx.text.orEmpty
            .debounce(.milliseconds(300), scheduler: MainScheduler.instance)
            .distinctUntilChanged()
            .map(Event.searchChanged),
        me.tableView.rx.willDisplayCell
            .filter { $0.indexPath.row == state.results.count - 3 }
            .map { _ in Event.loadNextPage },
    ]
    return Bindings(subscriptions: subscriptions, events: events)
}

9.6 组装系统

let state$ = Driver.system(
    initialState: State.empty,
    reduce: State.reduce,
    feedback: uiFeedback, loadPage
)

三行代码组装一个完整的搜索系统： 状态管理 + 网络请求 + UI 绑定 + 分页加载。

10. 与其他架构的对比

10.1 核心差异表

特性	RxFeedback	Redux	Elm	MVVM
单向数据流	✅	✅	✅	❌ (双向绑定)
纯函数 Reducer	✅	✅	✅	❌
副作用模型	Feedback Loop	Middleware	Cmd	ViewModel
反馈循环	✅ 显式	❌ (需插件)	✅ (Msg)	❌
平台限定	RxSwift	JavaScript	Elm 语言	各平台
核心代码量	~300行	~200行	编译器内置	无框架
循环依赖处理	原生支持	需要额外方案	TEA 模型	需手动管理

10.2 与 Redux 的详细对比

Redux 的中间件模式：

Action → Middleware₁ → Middleware₂ → Reducer → State
              ↕ (副作用)         ↕ (副作用)

Middleware 工作在 Action 经过 Reducer 之前
Middleware 之间是链式关系

RxFeedback 的反馈循环模式：

Event → reduce → State ─→ feedback₁ (副作用) ─→ Event
                 └─────→ feedback₂ (副作用) ─→ Event

feedback 观察的是 Reducer 之后的状态
feedback 之间是并行关系

核心差异：

Redux 中间件看到的是"事件"，RxFeedback 的 feedback 看到的是"状态"。这在设计哲学上导致本质区别 — feedback 可以做"当某个状态条件满足时执行副作用"，而 Redux 中间件更适合"当某个 action 发生时拦截并做额外处理"。

10.3 与 MVVM 的对比

MVVM	RxFeedback
ViewModel 同时负责业务逻辑和状态	reduce 只管状态，feedback 只管副作用
View ↔ ViewModel 双向绑定	严格的单向数据流
状态分散在 ViewModel 的属性中	状态集中在一个结构体中
副作用可以与业务逻辑混在一起	副作用完全隔离
测试需要 mock ViewModel	reduce 是纯函数，feedback 可独立测试

11. 设计亮点与技术总结

11.1 架构亮点

亮点	说明
极简核心	~300 行代码实现完整的架构框架，没有任何冗余
control theory in code	将控制论中的反馈循环概念直接映射为 `(Observable<State>) → Observable<Event>`
ReplaySubject 作为信号总线	用 `bufferSize: 1` 的 ReplaySubject 精巧地连接了状态输出和 feedback 输入
副作用完全隔离	`reduce` 是纯函数（同步、可测试），所有异步和副作用被隔离在 `feedback` 中
可组合的 feedback	多个 feedback 并行运行，互不干扰，各自独立产生事件
解决循环依赖	反馈循环天然建模循环依赖，无需额外机制
层级无关	同一套 `system` 可应用于 App 级、页面级、组件级
DI 天然适配	纯函数 reduce 和 feedback 都可以通过构造函数注入依赖
生命周期安全	dispose 时级联取消所有副作用，无内存泄漏

11.2 关键设计决策

1. 为什么用 `ReplaySubject(bufferSize: 1)` 而不是 `BehaviorRelay`？

BehaviorRelay 需要初始值，而 feedback 订阅时初始状态已经发出
ReplaySubject 让新 feedback 在添加时自动收到最新状态
bufferSize: 1 确保只有最新状态被保留，避免内存浪费

2. 为什么用 `flatMapLatest` 而不是 `flatMap`？

// react 中使用 flatMapLatest
state$.filter(request).flatMapLatest { effects($0) }

用户快速切换搜索词 → flatMap 会保留所有请求 → 竞态条件
flatMapLatest → 新状态到来自动取消旧请求 → 结果一定是最新的

3. 为什么 `reduce` 必须是同步纯函数？

状态变更必须是原子的 — 不允许异步更新某个字段
纯函数保证 scan 的结果可预测、可重现
纯函数不依赖外部环境，单元测试成本接近于零

11.3 适用场景

场景	适合？	原因
搜索/过滤功能	✅	清晰的状态机模型
表单验证	✅	状态集中管理，逻辑清晰
复杂动画	⚠️	帧级状态可能过度设计
简单列表展示	⚠️	MVC 可能更简单
实时数据同步	✅	多个 feedback 处理不同通道
AI 对话界面	✅	状态机+流式响应天然适合

11.4 关键文件清单

文件	内容
`Feedback.swift`	核心类型别名（Feedback, Query, Bindings）
`System.swift`	`system()` 操作符实现
`ObservableType+RxFeedback.swift`	`bind()` 和 `react()` 辅助方法
`ObservableSchedulerContext.swift`	调度器感知的绑定上下文
`RxFeedback+RxCocoa.swift`	RxCocoa 集成扩展

结束语

RxFeedback 证明了好的架构不需要复杂的代码。300 行的核心实现，一个 Feedback 类型别名，一个 system 操作符，就构成了一个功能完备的单向数据流系统。

它的精妙之处在于将控制论中"反馈循环"这一抽象概念直接映射为 RxSwift 的类型系统：(Observable<State>) -> Observable<Event>。这个简单的签名承载了架构的全部意图 — 状态驱动副作用，副作用产生事件，事件更新状态 — 循环往复，生生不息。

在 SwiftUI + Combine 大行其道的今天，RxFeedback 的设计思想依然闪耀。它所代表的 声明式、单向数据流、副作用隔离 理念，已经成为现代 UI 框架（SwiftUI, Jetpack Compose, Flutter）的共识。理解 RxFeedback，就是理解现代前端架构的基因。

参考资料

RxFeedback.swift 官方仓库 — NoTests/RxFeedback.swift，RxFeedback 源码、示例项目及 README 设计理念
RxSwift 中文文档 — 7.2 RxFeedback — RxFeedback 中文入门教程与核心概念讲解
RxSwift 中文文档 — 7. RxSwift 常用架构 — RxFeedback、ReactorKit 等 RxSwift 架构对比
RxSwift Community Projects — RxSwift 社区生态项目索引
RxFeedback 学习笔记 — 个人博客的 RxFeedback 实践总结
RxFeedback 架构解析 — 简书 — RxFeedback 入门分析文章
RxFeedback：简化 RxSwift 架构的利器 — CSDN — RxFeedback 框架介绍与实践
iOS Architecture 集合 — GitHub — 包含 RxFeedback 示例的 iOS 架构对比项目
Redux Observable — GitHub — 与 RxFeedback 设计理念相近的 Redux 中间件方案