使用BlockingCollection更新ObservableCollection

Question

我订阅了一个服务，当接收到新元素时，该服务会触发事件，我将此元素添加到一个BlockingCollection中。
我有一个第二个线程在运行，它将循环BlockingCollection，以将元素添加/更新到可观察集合中。

问题是如何在ObservableCollection中添加元素？ 我知道我不能仅仅在这种类型的集合上执行.add，因为它需要在UI线程上执行。所以我尝试使用不同的ObservableCollection子类，它们使用dispatcher来调度添加元素，但每一次都会遇到相同的错误

“An unhandled exception of type 'System.StackOverflowException' occurred in Unknown Module.”

故障排除提示如下：

"确保您没有无限循环或无限递归"

事实上，我确实有某种形式的无限循环，因为BlockingQueue一直在接收新元素，大约每秒5到10个左右。 如果我不将控件绑定到observablecollection，或者使用List而不是observablecollection，则不会出现异常。

Class ElementHolder
{
    private ExternalService _externalService;
    private ObservableCollection<Element> _elementsList = new ObservableCollection<Element>();
    private BlockingCollection<Element> _elementsReceived = new BlockingCollection<Element>();

    public ObservableCollection<Element> ElementsList
    {
        get
        {
            return _elementList;
        }
        set
        {
            _elementList = value;
        }
    }

public ElementHolder()
    {
        Task.Factory.StartNew(ReadElements);
        _externalService = new ExternalService();
        _externalService.ReceivedNewElement += new Action<Element>(o => _elementsReceived.Add(o));
        _externalService.Subscribe();
    }

private void ReadElements()
    {
        foreach (Element element in _elementsReceived.GetConsumingEnumerable())
        {
            Element item = _elementsList.FirstOrDefault(o => o.ID == element.ID);
            if (item == null)
            {
                _elementList.Add(element);
            }
            else
            {
                item.Update(element);
            }
        }
    }

编辑 当我追踪问题时，这个bug突然自己消失了。我尝试简化事情以便真正理解问题出在哪里，然后它开始工作了。当把事情重新组合起来时，它仍然可以工作...但是偶尔会出现一些看起来不相关的原因，比如向我的listview添加样式。我开始认为第三方dll存在问题。

- kstarbucks51

Answer 1

这是一个完美的例子，展示了反应式扩展（Reactive Extensions）是非常有用和巧妙的工具。使用它们需要掌握一定的学习曲线，但由于您在这里有一个特定的案例，我将插入反应式代码来实现您的目标，假设我正确理解了您的目标。

请注意，您需要安装反应式扩展，并且需要两个using语句（System.Reactive.Linq和System.Reactive.Subjects），以及引用System.Reactive和System.Reactive.Windows.Threading dlls。请参见MSDN上的反应式扩展。

class ElementHolder
{
    public ObservableCollection<Element> ElementsList { get; set; }
    private ExternalService _externalService = new ExternalService();
    private IDisposable _elementSubscription;
    private Subject<Element> _elementSubject = new Subject<Element>();

    public ElementHolder()
    {
        _externalService.ReceivedNewElement += _elementSubject.OnNext;
        _externalService.Subscribe();

        ElementList = new ObservableCollection<Element>();
        _elementSubscription = _externalService.ObserveOnDispatcher().Subscribe(NextElement);
    }

    private void NextElement(Element e)
    {
        Element item = ElementsList.FirstOrDefault(o => o.ID == element.ID);
        if (item == null) {
            _elementList.Add(element);
        }
        else {
            item.Update(element);
        }
    }
}

- theMayer

---

今天我其实快速浏览了一下 RX，并将 introtorx.com 加入了书签，因为它看起来非常有趣。 但不幸的是，在这种情况下，外部服务由第三方提供的 dll 提供，而我无法更改它。 - kstarbucks51

有一些方法可以使用响应式编程来实现这个，我会回复一些代码，但你可以使用FromEventPattern构造函数。或者只需将代码转移到您的类中，它仍然可以工作。 - theMayer

谢谢你的帮助，我现在没有时间测试它，因为我需要赶着完成这个项目，而且现在它已经可以工作了。但是我肯定会在周末仔细研究一下。 - kstarbucks51

Answer 2

答案中有一个小错误。请参见下面更正后的行：

_elementSubscription = _elementSubject.ObserveOnDispatcher().Subscribe(NextElement);

我花了一些时间才弄清楚这个问题，但rmayer06的答案解决了我的问题。如果可以评论答案，我会留言的。

- mobese46