r3-reactive-extensions

R3：面向 .NET 的现代响应式扩展

R3 是 Cysharp 从头重新实现的响应式扩展——“dotnet/reactive 和 UniRx 的新未来”。它保留了基于事件的 LINQ 编程模型，但重建了核心类型、错误约定和调度器，以解决 System.Reactive（Rx.NET）中长期存在的问题。在 C# 中组合事件流、UI 输入、计时器或基于推送的管道时，请使用此技能。

权威来源（在代码和文档中链接到以下内容）：

• 仓库：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

何时使用此技能

在以下情况下使用此技能：

• 组合随时间发生的事件——UI 输入、传感器／信息源更新、websocket 消息、领域事件
• 需要 debounce、throttle、merge、combine-latest、distinct-until-changed 等运算符
• 使用 ReactiveProperty / BindableReactiveProperty 构建 MVVM 状态
• 在基于推送的流与 Task / async 和 IAsyncEnumerable 之间进行桥接
• 从 System.Reactive、UniRx 或 IObservable<T> 代码迁移
• 遇到 Rx 的痛点：订阅因异常而终止、调度器开销过大或难以排查泄漏

不适合的场景： 请求／响应式 I/O（使用 async/await）、带背压的有界生产者／消费者（使用 System.Threading.Channels），或带批处理／背压的服务端流处理（使用 Akka.NET Streams）。和所有 Rx 一样，R3 是基于推送且没有背压的。选择这些方案时，请参阅 csharp-concurrency-patterns 技能。

参考文件

• rx-net-differences.md：与 System.Reactive（Rx.NET）的所有重要差异——新的核心类型、错误模型、运算符重命名、移除的 API、调度器替换，以及迁移检查清单。
• async-and-integration-patterns.md：常见模式——使用 AwaitOperation 进行异步分发、Task 集成、IAsyncEnumerable 往返转换、ReactiveProperty / MVVM、主题，以及订阅生命周期。
• scheduling-and-concurrency.md：R3 如何处理并发更新（线程约定、Synchronize、ObserveOn）、TimeProvider 与 FrameProvider 的区别及各自的适用场景，以及使用伪造 Provider 进行确定性测试。

本技能中的所有内容均已通过 R3 1.3.1 实证验证。参考文件中包含捕获的输出作为依据。

---

R3 存在的原因（“为什么使用它”）

作者 (neuecc) 构建 R3 是为了解决 System.Reactive 中的具体缺陷：

1. 异常会悄无声息地终止订阅。 在 Rx 中，管道里出现一个异常就会调用 OnError，并且永久取消订阅——对于长生命周期的事件流，这是“价值十亿美元的错误”（单个异常的 UI 事件就会拆除整个订阅）。R3 会将错误路由到 OnErrorResume，并且默认保持订阅活动。
2. IScheduler 过于笨重且令人困惑。 实测发现 ImmediateScheduler/Merge 会导致 服务器内存/CPU 开销显著增加。R3 移除了 IScheduler，改用 .NET 8 的 TimeProvider （墙上时钟）以及新的 FrameProvider（帧时钟）。
3. 订阅泄漏难以发现。 R3 将每个 Observable<T> 都设为抽象类，让所有订阅都汇集到同一处，从而使 ObservableTracker 能够列出每个仍存活的订阅及其堆栈跟踪。
4. Rx 与异步操作的结合不够自然。 R3 将 Rx 视为事件优先，并添加了明确的桥接接口（AwaitOperation、FromAsync、ToAsyncEnumerable），而不是假装事件是基于拉取的序列。
5. 一个库，适用于所有 UI。 跨平台核心加上精简的提供程序包，支持 Unity、Godot、WPF、WinForms、Avalonia、WinUI3、MAUI、Stride、MonoGame 和 Blazor。

---

安装

dotnet add package R3
# Platform glue (pick what applies): R3.WPF, R3.Avalonia, R3.WinForms, R3.Unity (UPM),
# R3.Godot, ObservableCollections.R3, etc. See the repo README for the full list.

using R3;

---

心智模型

R3 用抽象类替代 Rx 的接口，并用一个携带结果的完成通知，替代 Rx 的双方法错误约定。

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

public abstract class Observer<T> : IDisposable               // the observer IS the subscription
{
    public void OnNext(T value);
    public void OnErrorResume(Exception error);               // error WITHOUT unsubscribing
    public void OnCompleted(Result result);                   // success OR failure terminates
}

其语法为 (OnNext | OnErrorResume)* OnCompleted(Result)?。请注意这与 Rx 的 OnNext* (OnError | OnCompleted)? 不同：错误与终止彼此独立。错误只是一种通知；只有 OnCompleted 才会结束流，并且它携带一个 Result，其值为 Result.Success 或 Result.Failure(exception)。

快速入门

using R3;

var subscription = Observable
    .EveryValueChanged(model, m => m.SearchText)   // emits when the property changes
    .Debounce(TimeSpan.FromMilliseconds(300))      // Rx called this "Throttle" (see differences)
    .DistinctUntilChanged()
    .SubscribeAwait(async (text, ct) =>
    {
        var results = await _api.SearchAsync(text, ct);
        Render(results);
    }, AwaitOperation.Switch);                      // cancel the in-flight search on a new keystroke

// Dispose to unsubscribe; or route into a DisposableBag / AddTo(token).
subscription.Dispose();

---

核心行为，已验证

默认情况下，错误不会终止流

var subject = new Subject<int>();
subject.Select(x => 100 / x).Subscribe(
    onNext:        x => Console.WriteLine($"next {x}"),
    onErrorResume: e => Console.WriteLine($"errorResume {e.GetType().Name}"),
    onCompleted:   (Result r) => Console.WriteLine($"completed IsSuccess={r.IsSuccess}"));

subject.OnNext(2);   // next 50
subject.OnNext(0);   // errorResume DivideByZeroException   <-- NOT terminated
subject.OnNext(5);   // next 20                             <-- subscription is still alive!
subject.OnCompleted(); // completed IsSuccess=True

这是与 Rx 相比最大的行为变化。若要恢复传统的“错误会终止序列”行为，请插入 .OnErrorResumeAsFailure()——此时错误会传递到 OnCompleted(Result.Failure(e))，下游的 OnNext 也会停止。使用 Catch 恢复。完整的捕获运行记录以及（有意缺失的）Retry 说明见 rx-net-differences.md。

异步调度是显式的

R3 的异步运算符（SubscribeAwait、SelectAwait、WhereAwait，等等）接受一个 AwaitOperation，用于决定当值到达速度快于异步工作完成速度时该如何处理：

AwaitOperation	重叠行为	典型用途
Sequential（默认）	将值排入队列，逐个处理	按顺序处理
Drop	有任务正在运行时忽略新值	防抖提交／冷却
Switch	取消正在运行的任务，启动新任务	边输入边搜索、以最新值为准
Parallel	并发运行所有任务	独立扇出
SequentialParallel	并发运行，按顺序发出结果	并行映射、按序输出
ThrottleFirstLast	处理一组突发值中的第一个和最后一个	前沿／后沿采样

已验证这些行为完全符合描述（包括 Switch 取消被取代操作的 CancellationToken）。参见 async-and-integration-patterns.md。

Task 和 IAsyncEnumerable 桥接

// Task -> Observable
await Observable.FromAsync(async ct => await LoadAsync(ct)).FirstAsync();

// Observable -> Task (terminal operators return Task<T>)
List<int> all = await source.ToListAsync();
int last      = await source.LastAsync();

// IAsyncEnumerable -> Observable, and back
await asyncEnumerable.ToObservable().ForEachAsync(Handle);
await foreach (var x in source.ToAsyncEnumerable()) { /* ... */ }

全部验证可用。详情及完整的终止运算符列表见 async-and-integration-patterns.md。

---

R3 如何处理并发更新

R3 不会串行化并发生产者。 与 Rx 一样，它遵循 Rx 语法：不得从多个线程并发或重入调用 OnNext。运算符（Where、 Select、Subject，……）不会在内部加锁。同时从多个线程向有状态下游推送 OnNext 会破坏状态——测试中，向 List<T> 订阅者并发调用 OnNext 20,000 次，约有一半项目丢失，并且在运算符链中抛出了异常。

解决方法是明确设置边界：

// Multiple producer threads -> one serialized consumer
subject.Synchronize()                  // lock-based gate; delivery becomes single-threaded
       .Where(x => x.IsValid)
       .Subscribe(Handle);             // verified: 10000/10000 items, no corruption

// Or marshal onto a context/threadpool, which also serializes delivery:
source.ObserveOnThreadPool().Subscribe(Handle);

// For shared MVVM state written from many threads:
var counter = new SynchronizedReactiveProperty<int>(0);   // thread-safe writes

实用规则： 如果多个线程可能向流发布数据，请在源后立即添加 Synchronize()（或 ObserveOn*），或者使用 SynchronizedReactiveProperty。完整的竞态复现示例和输出见 scheduling-and-concurrency.md。

---

时间与帧：TimeProvider 和 FrameProvider

R3 有两种“时间”概念，两者都是可以在测试中伪造的抽象：

• TimeProvider（.NET 8 BCL 类型）＝墙上时钟时间。供 Delay、Debounce、 Interval、Timer、Timeout 使用。这是服务器和业务代码使用的时间。
• FrameProvider（R3 专用）＝帧时钟。供 EveryUpdate、DelayFrame(n)、 IntervalFrame(n) 等使用。

何时需要 FrameProvider？ 当“进度”是按渲染／更新帧数而非经过时间衡量时：

• 游戏引擎（Unity、Godot、Stride、MonoGame）——逻辑按引擎的更新循环逐帧运行， 因此会遵循暂停和时间缩放，并与渲染保持同步。
• UI 渲染循环（WPF/Avalonia/WinUI 合成帧）——逐帧响应。
• 确定性测试——FakeFrameProvider.Advance(n) 无需经过实际时间即可推进帧， 正如 FakeTimeProvider.Advance(timeSpan) 推进时钟一样。

普通的服务器和业务代码几乎从不需要 FrameProvider——那是 TimeProvider 的 适用范围。两种伪造提供程序都能让依赖时间的管道完全确定；示例见 scheduling-and-concurrency.md。

---

最佳实践摘要

应该做

• 将 OnErrorResume 作为默认选择：设计能够承受单个错误事件的流。
• 只有确实希望错误终止流时，才添加 .OnErrorResumeAsFailure()。
• 为每个异步运算符有意识地选择 AwaitOperation（最新结果优先用 Switch， 按顺序处理用 Sequential，冷却期用 Drop）。
• 在任何可能接收多个线程发布数据的源后添加 Synchronize() / ObserveOn*。
• 将 TimeProvider 传给时间运算符，将 FrameProvider 传给帧运算符，以便测试使用 FakeTimeProvider / FakeFrameProvider。
• 管理生命周期：将订阅存入 DisposableBag、CompositeDisposable，或调用 .AddTo(cancellationToken)；在开发环境中启用 ObservableTracker 以发现泄漏。
• 使用 ReactiveProperty 表示去重后的可观察状态；使用 BindableReactiveProperty 表示 绑定到 XAML 的状态。

不应该做

• 不要假设异常会终止流（这是 Rx 的行为，不是 R3）。
• 不要使用 R3 已重命名的 Rx 名称：这里是 Debounce（不是 Throttle）、ThrottleLast （不是 Sample）、Chunk（不是 Buffer）。1.3.1 中不包含 Retry、GroupBy、Finally 和普通的 Buffer——替代方案见差异文件。
• 不要在未使用 Synchronize() 的情况下，从多个线程并发／重入调用 OnNext。
• 不要将 R3 用于需要背压的高吞吐量管道——请使用 Channels 或 Akka.NET Streams。
• 不要阻塞等待终结运算符（.Result／.Wait()）；它们返回 Task<T>——请使用 await。

---

其他资源

• R3 仓库： https://github.com/Cysharp/R3
• 完整 README / 操作符参考： https://github.com/Cysharp/R3/blob/main/README.md
• 设计原理（neuecc）： 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
• 相关技能： csharp-concurrency-patterns（选择 R3、async/await、Channels 还是 Akka.NET）