r3-reactive-extensions

用途

R3 是 Cysharp 对 Reactive Extensions 的从头重写，被称为「dotnet/reactive 与 UniRx 的新未来」。它保留了 LINQ-over-events 的编程模型，但重建了核心类型、错误契约与调度器，以修复 System.Reactive (Rx.NET) 长期存在的问题。本技能用于在 C# 中组合事件流、UI 输入、定时器或推送式管道。

主要能力

核心模型：R3 用抽象类替换 Rx 的接口，用携带结果的单一完成事件替换 Rx 的双方法错误契约。

public abstract class Observable<T>
{
    public IDisposable Subscribe(Observer<T> observer);
    protected abstract IDisposable SubscribeCore(Observer<T> observer);
}

public abstract class Observer<T> : IDisposable
{
    public void OnNext(T value);
    public void OnErrorResume(Exception error);
    public void OnCompleted(Result result);
}

语法为 (OnNext | OnErrorResume)* OnCompleted(Result)?。错误只是一个通知，不会终止流；只有 OnCompleted 结束流，并携带 Result.Success 或 Result.Failure(exception)。

错误默认不终止：一个异常不会像 Rx 那样调用 OnError 并永久退订。如需恢复经典的「错误终止序列」行为，可插入 .OnErrorResumeAsFailure()，并用 Catch 恢复。
异步调度显式化：异步操作符（SubscribeAwait、SelectAwait、WhereAwait 等）接受 AwaitOperation 参数，决定值到达速度快于异步完成时的行为：Sequential（默认，排队逐个执行）、Drop（运行期间忽略新值）、Switch（取消运行中的操作并启动新的）、Parallel（全部并发）、SequentialParallel（并发但按序输出）、ThrottleFirstLast（突发取首尾）。
Task 与 IAsyncEnumerable 桥接：Observable.FromAsync 将 Task 转为 Observable，终结操作符（ToListAsync、LastAsync 等）返回 Task<T>，ToObservable/ToAsyncEnumerable 支持双向转换。
并发更新处理：R3 不序列化并发生产者，假设 OnNext 不被多线程并发或重入调用。多线程发布时应使用 Synchronize() 或 ObserveOn*，共享 MVVM 状态使用 SynchronizedReactiveProperty<int>。
时间与帧：使用 .NET 8 的 TimeProvider（墙钟时间，用于 Delay、Debounce、Interval、Timer、Timeout）和 R3 特有的 FrameProvider（帧时钟，用于 EveryUpdate、DelayFrame(n)、IntervalFrame(n)），两者均可在测试中用 FakeTimeProvider / FakeFrameProvider 驱动。
订阅生命周期：所有 Observable<T> 均为抽象类，订阅集中管理，可用 ObservableTracker 列出所有存活订阅及堆栈跟踪，便于发现泄漏。

使用方式

安装：

dotnet add package R3

平台胶水包按需选择：R3.WPF、R3.Avalonia、R3.WinForms、R3.Unity (UPM)、R3.Godot、ObservableCollections.R3 等。

using R3;

快速开始示例：

using R3;

var subscription = Observable
    .EveryValueChanged(model, m => m.SearchText)
    .Debounce(TimeSpan.FromMilliseconds(300))
    .DistinctUntilChanged()
    .SubscribeAwait(async (text, ct) =>
    {
        var results = await _api.SearchAsync(text, ct);
        Render(results);
    }, AwaitOperation.Switch);

subscription.Dispose();

管理生命周期：将订阅放入 DisposableBag、CompositeDisposable 或调用 .AddTo(cancellationToken)；开发时开启 ObservableTracker 捕获泄漏。使用 ReactiveProperty 表示去重的可观察状态，BindableReactiveProperty 表示 XAML 绑定状态。

适用场景

随时间组合事件：UI 输入、传感器/数据源更新、websocket 消息、领域事件
需要 debounce、throttle、merge、combine-latest、distinct-until-changed 等操作符
用 ReactiveProperty / BindableReactiveProperty 构建 MVVM 状态
桥接推送式流与 Task / async 及 IAsyncEnumerable
从 System.Reactive、UniRx 或 IObservable<T> 代码迁移
遇到 Rx 痛点：异常导致订阅死亡、调度器开销、泄漏排查
游戏引擎与 UI 渲染循环中按帧推进的逻辑

限制

不适用于请求/响应式 I/O（应使用 async/await）、带背压的有界生产者/消费者（应使用 System.Threading.Channels）、或带批处理/背压的服务端流处理（应使用 Akka.NET Streams）。R3 与所有 Rx 一样是推送式且无背压。
不要假设异常会结束流（那是 Rx，不是 R3）。
注意操作符改名的差异：是 Debounce（不是 Throttle）、ThrottleLast（不是 Sample）、Chunk（不是 Buffer）。在 1.3.1 中 Retry、GroupBy、Finally 和普通 Buffer 不存在。
不要在未使用 Synchronize() 的情况下从多线程并发/重入调用 OnNext。
不要在终结操作符上阻塞（.Result/.Wait()），它们返回 Task<T>，应 await。

参考与规范来源

仓库：https://github.com/Cysharp/R3
README（完整操作符参考）：https://github.com/Cysharp/R3/blob/main/README.md
作者设计理由：https://neuecc.medium.com/r3-a-new-modern-reimplementation-of-reactive-extensions-for-c-cf29abcc5826
NuGet：https://www.nuget.org/packages/R3
TimeProvider (BCL)：https://learn.microsoft.com/en-us/dotnet/api/system.timeprovider
FakeTimeProvider：https://www.nuget.org/packages/Microsoft.Extensions.TimeProvider.Testing

本技能内容已针对 R3 1.3.1 进行经验性验证。相关技能：csharp-concurrency-patterns（在 R3、async/await、Channels 与 Akka.NET 之间做选择）。

技能介绍

用途

主要能力

使用方式

适用场景

限制

参考与规范来源