关于c#：Observable.Repeat 势不可挡，是bug还是特性？

The Observable.Repeat is unstoppable, is it a bug or a feature?

本问题已经有最佳答案，请猛点这里访问。

当源 observable 的通知是同步的时，我注意到 Repeat 运算符的行为有些奇怪。生成的 observable 不能用后续的 TakeWhile 操作符停止，并且显然会永远继续运行。为了演示，我创建了一个源 observable，它产生一个值，它在每次订阅时递增。第一个订阅者获得值 1，第二个获得值 2 依此类推：

1
2
3
4
5
6
7
8
9
10
11

int incrementalValue = 0;
var incremental = Observable.Create<int>(async o =>
{
await Task.CompletedTask;
//await Task.Yield();

Thread.Sleep(100);
var value = Interlocked.Increment(ref incrementalValue);
o.OnNext(value);
o.OnCompleted();
});

然后我将运算符 Repeat、TakeWhile 和 LastAsync 附加到这个 observable 上，这样程序就会等到组合后的 observable 产生它的最后一个值：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

incremental.Repeat()
.Do(new CustomObserver(“Checkpoint A”))
.TakeWhile(item => item <= 5)
.Do(new CustomObserver(“Checkpoint B”))
.LastAsync()
.Do(new CustomObserver(“Checkpoint C”))
.Wait();
Console.WriteLine($“Done”);

class CustomObserver : IObserver<int>
{
private readonly string _name;
public CustomObserver(string name) => _name = name;
public void OnNext(int value) => Console.WriteLine($“{_name}: {value}”);
public void OnError(Exception ex) => Console.WriteLine($“{_name}: {ex.Message}”);
public void OnCompleted() => Console.WriteLine($“{_name}: Completed”);
}

这是这个程序的输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

Checkpoint A: 1
Checkpoint B: 1
Checkpoint A: 2
Checkpoint B: 2
Checkpoint A: 3
Checkpoint B: 3
Checkpoint A: 4
Checkpoint B: 4
Checkpoint A: 5
Checkpoint B: 5
Checkpoint A: 6
Checkpoint B: Completed
Checkpoint C: 5
Checkpoint C: Completed
Checkpoint A: 7
Checkpoint A: 8
Checkpoint A: 9
Checkpoint A: 10
Checkpoint A: 11
Checkpoint A: 12
Checkpoint A: 13
Checkpoint A: 14
Checkpoint A: 15
Checkpoint A: 16
Checkpoint A: 17
…

它永远不会结束！虽然 LastAsync 已经产生了它的值并完成了，但 Repeat 操作符仍在旋转！

只有当源 observable 同步通知其订阅者时才会发生这种情况。例如，取消注释行 //await Task.Yield(); 后，程序的行为与预期相同：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Repeat 操作符停止旋转，尽管它没有报告完成(我的猜测是它已被取消订阅)。

有没有什么方法可以使 Repeat 操作符的行为保持一致，而不管它接收到的通知类型(同步还是异步)？

.NET Core 3.0、C# 8、System.Reactive 4.3.2、控制台应用程序

相关讨论

我一直说 Observable.Create 很糟糕……
好吧，它更多地与 Scheduler.Immediate 有关。将调度程序更改为 incremental.ObserveOn(Scheduler.Default).Repeat() 并查看会发生什么。
我怀疑你锁定了取消订阅所需的线程。
@Enigmativity 它与 Observable.Create 无关。此实现也存在问题：var incremental = Observable.Defer(() => Observable.Return(++incrementalValue));
你是对的。然而 Observable.Create 使得创建具有此问题的可观察对象变得太容易了。
@Enigmativity with ObserveOn(Scheduler.Default) 通知是从各种线程接收的，本质上是模仿我在生产者内部使用 await Task.Yield() 的测试。但是我检查了另一个选项： ObserveOn(Scheduler.CurrentThread) 似乎可以彻底解决问题！使用此选项，程序按预期工作，并且一切都发生在一个线程中。 ObserveOn(Scheduler.CurrentThread) 有什么我应该注意的缺点吗？
在Repeat之前用SubscribeOn(Scheduler.CurrentThread)也解决了这个问题，天知道为什么！

从内存 Scheduler.CurrentThread 使用蹦床在当前线程(即当前上下文)上进行调度，因此它不会立即运行并取消阻塞线程。否则，使用 Scheduler.Immediate 您将面临死锁的风险。
@Enigmativity 是我的猜测，但是控制台应用程序没有同步上下文，因此令人费解。
您应该查看 System.Reactive.Concurrency 命名空间的源代码。那会让你的眼睛流泪。他们在那里做了很多技巧来使 Rx 工作。那里有一个同步上下文。

您可能期望 Repeat 的实现以 OnCompleted 通知为特色，但事实证明它是根据 Concat 实现的 – 无限流。

1
2
3
4
5
6
7
8
9
10

public static IObservable<TSource> Repeat<TSource>(this IObservable<TSource> source) =>
RepeatInfinite(source).Concat();

private static IEnumerable< T > RepeatInfinite< T >(T value)
{
while (true)
{
yield return value;
}
}

将责任转移到 Concat – 我们可以创建一个简化版本(血淋淋的实现细节在 TailRecursiveSink.cs 中)。除非 await Task.Yield().

提供了不同的执行上下文，否则它仍然会继续旋转

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

public static IObservable< T > ConcatEx< T >(this IEnumerable<IObservable< T >> enumerable) =>
Observable.Create< T >(observer =>
{
var check = new BooleanDisposable();

IDisposable loopRec(IScheduler inner, IEnumerator<IObservable< T >> enumerator)
{
if (check.IsDisposed)
return Disposable.Empty;

if (enumerator.MoveNext()) //this never returns false
return enumerator.Current.Subscribe(
observer.OnNext,
() => inner.Schedule(enumerator, loopRec) //<– starts next immediately
);
else
return inner.Schedule(observer.OnCompleted); //this never runs
}

Scheduler.Immediate.Schedule(enumerable.GetEnumerator(), loopRec); //this runs forever
return check;
});

作为一个无限流，enumerator.MoveNext() 总是返回 true，所以另一个分支永远不会运行 – 这是预期的；这不是我们的问题。

当 o.OnCompleted() 被调用时，它立即调度下一个迭代循环
Schedule(enumerator, loopRec) 同步调用下一个 o.OnCompleted()，并且它无限地继续 – 没有一点可以逃脱这个递归。

如果你有一个带有 await Task.Yield() 的上下文切换，那么 Schedule(enumerator, loopRec) 会立即退出，并且 o.OnCompleted() 会被非同步调用。

Repeat 和 Concat 使用当前线程在不改变上下文的情况下工作 – 这不是不正确的行为，但是当同样的上下文也用于推送通知时，它可能导致死锁或被陷入
永久蹦床。

带注释的调用栈

1
2
3
4
5
6
7
8
9
10
11
12
13
14

[External Code]
Main.AnonymousMethod__0(o) //o.OnCompleted();
[External Code]
ConcatEx.__loopRec|1(inner, enumerator) //return enumerator.Current.Subscribe(…)
[External Code]
ConcatEx.AnonymousMethod__2() //inner.Schedule(enumerator, loopRec)
[External Code]
Main.AnonymousMethod__0(o) //o.OnCompleted();
[External Code]
ConcatEx.__loopRec|1(inner, enumerator) //return enumerator.Current.Subscribe(…)
[External Code]
ConcatEx.AnonymousMethod__0(observer) //Scheduler.Immediate.Schedule(…)
[External Code]
Main(args) //incremental.RepeatEx()…

The Observable.Repeat is unstoppable, is it a bug or a feature?

猜你喜欢