System.Reactive 6.0.1

Rx (.NET 的响应式扩展)

Rx 允许使用可组合、声明性模型进行事件驱动的编程。

入门教程

在命令行中运行以下命令

mkdir TryRx
cd TryRx
dotnet new console
dotnet add package System.Reactive

或者，如果您已安装 Visual Studio，创建一个新的 .NET 控制台项目，然后使用 NuGet 软件包管理器将对 System.Reactive 的引用添加到项目中。

然后您可以将此代码添加到您的 Program.cs 中。这创建了一个可观察的源（ticks），每秒钟产生一个事件。它还添加了一个处理程序到该源，为每个事件在控制台写入一条消息。

using System.Reactive.Linq;

IObservable<long> ticks = Observable.Timer(
    dueTime: TimeSpan.Zero,
    period: TimeSpan.FromSeconds(1));

ticks.Subscribe(
    tick => Console.WriteLine($"Tick {tick}"));

Console.ReadLine();

示例

将现有事件源包装为 Rx IObservable<T>

如果您有一个不支持 Rx 直接但提供 .NET 事件的现有事件源，则可以使用 Observable.FromEventPattern 方法将其带入 Rx 世界。

using System.Reactive.Linq;

FileSystemWatcher fsw = new FileSystemWatcher(@"C:\temp");

IObservable<FileSystemEventArgs> changeEvents = Observable
    .FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
        h => fsw.Changed += h,
        h => fsw.Changed -= h)
    .Select(e => e.EventArgs);

fsw.EnableRaisingEvents = true;

等待不活动状态

有时在采取行动之前等待不活动期间可能很有用。例如，如果您有监控文件系统中的一个目录的代码，处理修改或添加的文件，活动频繁出现是很常见的。例如，如果用户正在将多个文件复制到您正在观察的文件夹中，将会有多个更改，并且等待这些更改停止然后一次性处理所有更改可能更有效，而不是立即尝试处理一切。

此示例定义了一个可以附加到任何源的 Rx 操作符。它会等待该源开始产生事件，然后等待它再次停止指定的时间。每次发生这种情况时，它都会报告在最后一个不活动期间发生的所有活动。

static class RxExt
{
    public static IObservable<IList<T>> Quiescent<T>(
        this IObservable<T> src,
        TimeSpan minimumInactivityPeriod,
        IScheduler scheduler)
    {
        IObservable<int> onoffs =
            from _ in src
            from delta in Observable.Return(1, scheduler).Concat(Observable.Return(-1, scheduler).Delay(minimumInactivityPeriod, scheduler))
            select delta;
        IObservable<int> outstanding = onoffs.Scan(0, (total, delta) => total + delta);
        IObservable<int> zeroCrossings = outstanding.Where(total => total == 0);
        return src.Buffer(zeroCrossings);
    }
}

（这是通过创建一个序列（onoffs）来实现的，每次活动发生时产生一个值为 1，然后在指定时间后产生一个相应的 -1。然后它使用 Scan 生成一个 outstanding 序列，它只是那些 onoffs 的累积总额。这实际上是对最近发生的事件数量的计数（其中“最近”定义为“小于 minimumInactivityPeriod 之前”。每次发生新事件时，运行总额增加 1，但是每当时限经过，它就减去 1。因此，当这个值降到 0 时，表示没有在 minimumInactivityPeriod 内发生任何事件。 zeroCrossings 序列仅选取那些 outstanding 降为 0 的事件。这导致每当有一些活动后跟随 minimumInactivityPeriod 的不活动时，zeroCrossings 就引发一个事件。最后，我们将这个结果连接到 Buffer 操作符，它将输入事件（src）切片。通过传递 zeroCrossings 源，我们告诉 Buffer 在源不活动时分配一个新切片。结果是 Quiescent 返回的源在没有活动后跟随指定的不活动期之前不做任何事情，在这一点上，它生成一个事件，报告自上次不活动期间或初次情况以来的所有源事件。）

您可以将此与前面的示例中更新的 FileSystemWatcher 一起使用。

IObservable<IList<FileSystemEventArgs>> fileActivityStopped = changeEvents
    .Quiescent(TimeSpan.FromSeconds(2), Scheduler.Default);

await fileActivityStopped.ForEachAsync(
    a => Console.WriteLine($"File changes stopped after {a.Count} changes"));

（注意：这仅使用 Changed 事件。实际应用程序可能还需要查看 FileSystemWatcher 的 Created、Renamed 和 Deleted 事件。）

反馈

您可以在https://github.com/dotnet/reactive 存储库中创建问题。