抛出可观察到的 LINQ 不一致异常

Question

在编写股票市场交易员IObserver 的过程中，我遇到了三个错误，主要来自Reactive Extensions 库。

我有以下CompanyInfo 类：

public class CompanyInfo
{
    public string Name { get; set; }

    public double Value { get; set; }
}

还有一个叫做StockMarket的IObservable<CompanyInfo>：

public class StockMarket : IObservable<CompanyInfo>

我的Observer 如下所示：

public class StockTrader : IObserver<CompanyInfo>
{
    public void OnCompleted()
    {
        Console.WriteLine("Market Closed");
    }

    public void OnError(Exception error)
    {
        Console.WriteLine(error);
    }

    public void OnNext(CompanyInfo value)
    {
        WriteStock(value);
    }

    private void WriteStock(CompanyInfo value) { ... }
}

我运行以下代码：

StockMarket market = GetStockMarket();
StockTrader trader = new StockTrader();

IObservable<CompanyInfo> differential = market  //[F, 1], [S, 5], [S, 4], [F, 2]
    .GroupBy(x => x.Name)                       //[F, 1], [F, 2]; [S, 5], [S, 4]
    .SelectMany(x => x                  //4, 8, 2, 3
        .Buffer(2, 1)                   //(4, 8), (8, 2), (2, 3), (3)
        .SkipLast(1)                    //(4, 8), (8, 2), (2, 3)
        .Select(y => new CompanyInfo    //(+100%), (-75%), (+50%)
        {
            Name = x.Key,
            Value = (y[1].Value - y[0].Value) / y[0].Value
        })                                      //[F, +100%]; [S, -20%]
    );

using (IDisposable subscription = differential.Subscribe(trader))
{
    Observable.Wait(market);
}

出现三个错误之一：

• 从Reactive Extensions 中抛出以下ArgumentException：

  System.ArgumentException：已添加具有相同键的项目。 在 System.ThrowHelper.ThrowArgumentException（ExceptionResource 资源） 在 System.Collections.Generic.Dictionary`2.Insert(TKey key, TValue value, Boolean add) 在 System.Reactive.Linq.Observable.GroupBy'3._.OnNext(TSource 值)

• 以下IndexOutOfRangeException：

  参数名称：索引 在 System.ThrowHelper.ThrowArgumentOutOfRangeException（ExceptionArgument 参数，ExceptionResource 资源） 在 System.Collections.Generic.List'1.get_Item（Int32 索引） 在 StockMarketTests.c__DisplayClass0_0.b__2(IList'1 y) 在 System.Reactive.Linq.Observable.Select'2._.OnNext(TSource 值)

• Console 的文字偶尔会发生调整（颜色应该一致）：

什么会导致这些奇怪的症状？

Answer 1

Reactive Extensions 概念的最大优点之一是能够订阅“某处”发生的“发生” (IObservable) 和在这个'发生'上应用面向对象的概念——这不需要知道那个'somewhere'在哪里。

这种方式Reactive Extensions 大大简化了面向event 和producer-consumer problems 的编程。

订阅IObservable 而不知道观察到的数据来源的能力迫使订阅者假设通知是不可预测的。换句话说，当观察IObservable 你应该假设通知可以同时发送。

_{由于Reactive Externsions 的行为契约，IObservables 应该一次生产一个项目。通常情况下会发生这种情况，但有时外部实现不遵循该约定。}

让我们来看看这三个问题中的每一个：

GroupBy 不是线程安全的

---

GroupBy 通过返回一个IObservable<IGroupedObservable<T>> 工作，它的OnNext 方法调用外部IObservable 的OnNext 与匹配当前通知的IGroupedObservable<T>。它通过为Dictionary 中的每个键保留一个IGroupedObservable<T> _{（更准确地说是一个Subject<T>）} 来实现这一点 - 这并不奇怪 - 不是ConcurrentDictionary em>。这意味着两个邻近的通知可能会导致重复插入。

Select 并不孤单

---

Select 的线程安全由其提供的委托决定。在上面的例子中，提供给Select 方法的委托依赖于Buffer(2, 1) 将提供一个大小为2 的列表。Buffer 包含一个Queue，它不是并发，因此当从多个线程迭代时 - Buffer 的 Queue 可以为我们提供一些意想不到的结果。

另一个Exception 可能出于同样的原因抛出，如果y 将被提供null，则NullReferenceException，或者Queue 的InvalidOperationException 可以在迭代时被修改。

即使是基本的观察也不安全

---

最后但同样重要的是，即使您只进行基本观察，StockTrader 的OnNext 方法也会在非atomic operation 中修改控制台，这会导致奇怪的文本布局。

那你能做什么？

---

Synchronize 方法的存在使您能够验证您正在订阅 linear IObservable<T>，这意味着 对 OnNext 方法的调用不超过一次可以同时发生。

由于即使是GroupBy 扩展方法也不是线程安全的，所以Synchronize 方法需要在链的开头调用：

IObservable<CompanyInfo> differential = market  //[F, 1], [S, 5], [S, 4], [F, 2]
    .Synchronize()
    .GroupBy(x => x.Name)                       //[F, 1], [F, 2]; [S, 5], [S, 4]
    .SelectMany(x => x                  //4, 8, 2, 3
        .Buffer(2, 1)                   //(4, 8), (8, 2), (2, 3), (3)
        .SkipLast(1)                    //(4, 8), (8, 2), (2, 3)
        .Select(y => new CompanyInfo    //(+100%), (-75%), (+50%)
        {
            Name = x.Key,
            Value = (y[1].Value - y[0].Value) / y[0].Value
        })                                      
    );                                          //[F, +100%]; [S, -20%]

请注意，Synchronize 为您的查询添加了另一个代理 Observable，因此它会使查询慢一点，因此 您应该避免在不需要时使用它 。

Answer 2

您的代码的问题不在于查询，也不在于 Rx 本身。问题可能来自您实际的StockMarket 或StockTrader 实现。

现在，问题很可能是因为您正在为您的 market observable 创建两个订阅。

当你写这个时：

using (IDisposable subscription = differential.Subscribe(trader))
{
    Observable.Wait(market);
}

...您将获得两个market 的订阅。一个在differential.Subscribe(trader)，另一个因为Observable.Wait(market);。

我怀疑两个并发订阅导致了您的问题，但没有看到 StockMarket 的实现，我们无法判断它为什么会抛出。

这是实现您自己的 observable 和observer 实现的危险。你应该避免这样做。最好有一个属性 IObservable<CompanyInfo> CompanyValues { get; } 悬挂在使用标准 Rx 运算符构建的 CompanyInfo 之外。

而且您应该始终避免像.Wait(...) 这样的阻塞操作。

作为一项快速测试，我会将您当前的Observable.Wait(market); 替换为Thread.Sleep(?)，并具有足够长的睡眠时间，以查看您的代码是否正常运行。当然，您需要确保在后台调度程序上生成值（例如 Scheduler.Default）。

我运行了这段代码来测试你的查询：

public class CompanyInfo
{
    public string Name { get; set; }

    public double Value { get; set; }
}

public class StockTrader : IObserver<CompanyInfo>
{
    public void OnCompleted()
    {
        Console.WriteLine("Market Closed");
    }

    public void OnError(Exception error)
    {
        Console.WriteLine(error);
    }

    public void OnNext(CompanyInfo value)
    {
        WriteStock(value);
    }

    private void WriteStock(CompanyInfo value) { Console.WriteLine($"{value.Name} = {value.Value}"); }
}

public class StockMarket : IObservable<CompanyInfo>
{
    private CompanyInfo[] _values = new CompanyInfo[]
    {
        new CompanyInfo() { Name = "F", Value = 1 },
        new CompanyInfo() { Name = "S", Value = 5 },
        new CompanyInfo() { Name = "S", Value = 4 },
        new CompanyInfo() { Name = "F", Value = 2 },
    };

    public IDisposable Subscribe(IObserver<CompanyInfo> observable)
    {
        return _values.ToObservable().ObserveOn(Scheduler.Default).Subscribe(observable);
    }
}

...用这个：

StockMarket market = new StockMarket();
StockTrader trader = new StockTrader();

IObservable<CompanyInfo> differential = market  //[F, 1], [S, 5], [S, 4], [F, 2]
    .GroupBy(x => x.Name)                       //[F, 1], [F, 2]; [S, 5], [S, 4]
    .SelectMany(x => x                  //4, 8, 2, 3
        .Buffer(2, 1)                   //(4, 8), (8, 2), (2, 3), (3)
        .SkipLast(1)                    //(4, 8), (8, 2), (2, 3)
        .Select(y => new CompanyInfo    //(+100%), (-75%), (+50%)
        {
            Name = x.Key,
            Value = (y[1].Value - y[0].Value) / y[0].Value
        })                                      //[F, +100%]; [S, -20%]
    );

IDisposable subscription = differential.Subscribe(trader);

Thread.Sleep(10000);

我没有一次让它崩溃或导致任何异常。