[开发工具]C# 入门教程Subject发布和订阅

Reactive Extensions for .Net 为开发人员提供了一组功能，用于为 .Net 开发人员实现反应式编程模型，使用声明性操作使事件处理更简单、更具表现力。虽然反应式扩展的关键基石是 IObserver 和 IObservable 接口，但作为开发人员，您通常不需要自己实现这些接口。该库支持内置类型?Subject<T>?，它实现了两个接口并支持许多功能。
主题是库中不同可用主题的基础，还有其他主题 -?ReplaySubject<T>、?BehaviorSubject<T>?和?AsyncSubject<T>。了解它们之间的本质区别以及如何使用它们来更好地利用库是非常有用的。
在本文中，我们将比较?Subject<T>?和它的兄弟，以试图说明它们行为之间的差异。
主题<T>
如前所述，?Subject<T>?是可用主题的基础，它提供了一种使用库的简单方法，而无需自己实现?IObservable<T>?和?IObserver<T>?接口。Subject 类型的简单演示如下所示。

var subject = new Subject<int>();
subject.Subscribe(Console.Write);
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);

在上面的代码中，我们创建了一个?Subject<T>的实例， 并且由于它实现了?IObserver<T>?和?IObserverable<T>，因此使用了相同的实例来订阅和发布值到 IObserver。这里要注意的另一个重点是我们如何使用 接收动作作为输入的订阅方法的重载。将对发布的每个值执行该操作，在本例中是将数字打印到控制台。
让我们尝试在下图中显示已发布的值和由 IObserver（在此?Action<T>中）打印到控制台的值。这将帮助我们轻松比较剩余的兄弟姐妹和变体。

第一行表示发布的值，而第二行表示 IObserver 接收到的值。此外，我们添加了一行来指示 Observer 在什么执行点订阅数据流。这条线由垂直虚线表示。
在上面的代码中，我们注意到观察者在第一个值发布之前订阅了数据流。这在图像中进行了说明，因为?用户?线放置在第一个元素之前。如输出行所示，这对输出没有影响（此时）。
但是如果观察者只在一些值已经发布之后才订阅数据呢？这会对观察者收到的数据产生影响吗？在查看输出之前，让我们先编写相同的代码。

var subject = new Subject<int>();
subject.OnNext(1);
subject.OnNext(2);
subject.Subscribe(Console.Write);
subject.OnNext(3);
subject.OnNext(4);

在上面显示的代码中，可以观察到 Observer 仅在发布了两个值（1 和 2）之后才订阅数据流。正如人们所预料的那样，这将导致 Observer 无法接收在调用 subscribe 方法之前发布的数据。如下图所示。

如果您想读取所有已发布的值，即使观察者订阅晚了怎么办。这就是?ReplaySubject<T> 发挥作用的地方 。
重播主题<T>
ReplaySubject<T>?缓存这些值并为迟到的订阅者重播它们。这对于避免竞争条件很有用。让我们更改之前的代码以使用?ReplaySubject<T>?并看看它如何影响观察者接收到的内容。

var subject = new ReplaySubject<int>();
subject.OnNext(1);
subject.OnNext(2);
subject.Subscribe(Console.Write);
subject.OnNext(3);
subject.OnNext(4);

如上面的代码所示，代码几乎没有变化，除了我们现在使用?ReplaySubject<T>?而不是?Subject<T>。下图说明了对 Observer 接收到的数据的影响。

如图所示，即使订阅晚了，缓存的值现在也会重播给订阅者。当然，这个有用的功能需要自费。此实现将缓存订阅者发布的每个值，当数据量明显更大时，这可能会导致不良的内存问题。
但是，?ReplaySubject<T>?以不止一种方式解决了该问题。为了这个示例，我们将研究两个示例，它们将使用大小和时间约束来限制缓存值。
作为第一种情况，我们将使用缓存的大小来限制缓存的值。ReplaySubject<T>的构造函数 提供了一个重载，它接受一个表示缓存缓冲区大小（最大元素计数）的整数。在我们的示例中，让我们更改代码以将缓存大小限制为 1。

var subject = new ReplaySubject<int>(1);
subject.OnNext(1);
subject.OnNext(2);
subject.Subscribe(Console.Write);
subject.OnNext(3);
subject.OnNext(4);

请注意我们如何使用ReplaySubject<T>的构造函数重载将 Cache 的大小提供为 1 。这限制了缓存并确保仅缓存一个元素，并且一旦发布就会被新元素替换。更改的影响如下图所示。

限制缓存的另一种方法是限制缓存项目的时间，或者换句话说，为缓存的项目提供到期时间。
让我们编写代码来说明这个例子。

var subject = new ReplaySubject<int>(TimeSpan.FromMilliseconds(1000));
subject.OnNext(1);
Thread.Sleep(500);
subject.OnNext(2);
Thread.Sleep(200);
subject.OnNext(3);
Thread.Sleep(500);
subject.Subscribe(Console.Write);
subject.OnNext(4);
Thread.Sleep(500);

与前面的代码类似，我们使用了?ReplaySubject<T>?构造函数的重载来指定缓存中项目的到期时间。为了演示我们的案例，我们在值的发布之间引入了延迟。
由于 Observer 订阅前需要整整 1200 毫秒，因此任何超过 1000 毫秒到期持续时间的元素都将从缓存中删除。在此示例中，这将导致值 1 从缓存中删除，并且不会重播给迟到的订阅者。如下图所示。

还有其他重载?ReplaySubject<T>?可以提供更大的灵活性并微调缓存的值，但为了举例，我们将保留上面已经介绍过的两个示例。

行为主题<T>
BehaviorSubject<T>与ReplaySubject<T>?非常相似， 因为它有助于缓存值。但是有一个显着的区别。?BehaviorSubject<T>?仅缓存最后发布的值。在进一步讨论之前，让我们编写一些代码。

var subject = new BehaviorSubject<int>(0);
subject.OnNext(1);
subject.OnNext(2);
subject.Subscribe(Console.Write);
subject.OnNext(3);
subject.OnNext(4);

如果?BehaviorSubject<T>?只缓存一个值（最后知道的），那么它 与大小为 1的ReplaySubject<T>有何不同？如下所示的插图绝对反映了上面的代码。

然而，这并不完全正确。这里有两个显着的区别需要理解。第一个是默认值的存在。请注意，在上面的代码中，我们 在BehaviorSubject<T>的构造函数中提供了一个值0?作为默认值 。如果缓存中没有值（或者换句话说，在观察者订阅之前没有发布数据），那么将返回默认值。这 与大小为 1 的ReplaySubject<T>不同 ，后者没有任何值。此行为在以下序列的代码和可视化表示中得到了演示。

var subject = new BehaviorSubject<int>(0);
subject.Subscribe(Console.Write);
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);

第二个区别是?BehaviorSubject<T>?和?ReplaySubject<T>?在订阅已完成序列时的行为方式。BehaviorSubject<T>在 完成后订阅时不会有任何值，如下面的代码所示。

var subject = new BehaviorSubject<int>(0);
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);
subject.OnCompleted();
subject.Subscribe(Console.Write);

保证订阅者不会收到任何值，因为订阅发生在完成之后。

但是，?ReplaySubject<T>就是这种情况。无法保证 Observer 不会收到任何值，如下面的代码所示。

var subject = new ReplaySubject<int>(1);
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);
subject.OnCompleted();
subject.Subscribe(Console.Write);

如上面的代码所示，缓存的大小为 1，即使在完成调用之后调用订阅，缓存也会保留（直到满足到期条件），因此在这种情况下会收到最后发布的值。

异步主题<T>
AsyncSubject<T>是 我们将在本文中探讨的Subject<T>的最后一个兄弟姐妹， 与前两个（ReplaySubject 和 BehaviorSubject）非常相似，因为它也缓存了结果。但又一次出现了显着差异。仅当 序列被标记为完成时，?AsyncSubject<T>?才会发布最后一个缓存的值（它只缓存一个值，即最后一个）。
考虑以下代码。

var subject = new AsyncSubject<int>();
subject.Subscribe(Console.Write);
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);
subject.OnCompleted();

这将向观察者生成一个值，即在序列被标记为完成之前发布的最后一个值 - 值 4。如下图所示。

但是，如果我们跳过了将序列标记为完成的调用，会发生什么？让我们注释掉该行并重试。

var subject = new AsyncSubject<int>();
subject.Subscribe(Console.Write);
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnNext(4);

这不会为观察者生成任何数据，因为?AsyncSubject<T>?只会在序列被标记为完成后发布结果。

这是一个显着的区别，任何使用?AsyncSubject<T>的人 都应该记住这一点。
结论
本文演示了?Subject<T>的各种同级 及其一些变体之间的区别。了解这些细微差异通常很有用，因为如果您没有意识到它可能会展示与您期望的行为不同的行为。