快速执行且线程安全的可观察集合

Question

ObservableCollection 会针对对其执行的每个操作发出通知。首先，它们没有批量添加或删除调用，其次它们不是线程安全的。

这不是让他们变慢吗？我们不能有更快的替代方案吗？有人说包裹着 ObservableCollection 的 ICollectionView 速度很快？这种说法的真实性如何。

Answer 1

如果愿意的话，ObservableCollection 可以很快。 :-)

下面的代码是线程安全、更快的可观察集合的一个非常好的示例，您可以根据自己的意愿进一步扩展它。

using System.Collections.Specialized;

public class FastObservableCollection<T> : ObservableCollection<T>
{
    private readonly object locker = new object();

    /// <summary>
    /// This private variable holds the flag to
    /// turn on and off the collection changed notification.
    /// </summary>
    private bool suspendCollectionChangeNotification;

    /// <summary>
    /// Initializes a new instance of the FastObservableCollection class.
    /// </summary>
    public FastObservableCollection()
        : base()
    {
        this.suspendCollectionChangeNotification = false;
    }

    /// <summary>
    /// This event is overriden CollectionChanged event of the observable collection.
    /// </summary>
    public override event NotifyCollectionChangedEventHandler CollectionChanged;

    /// <summary>
    /// This method adds the given generic list of items
    /// as a range into current collection by casting them as type T.
    /// It then notifies once after all items are added.
    /// </summary>
    /// <param name="items">The source collection.</param>
    public void AddItems(IList<T> items)
    {
       lock(locker)
       {
          this.SuspendCollectionChangeNotification();
          foreach (var i in items)
          {
             InsertItem(Count, i);
          }
          this.NotifyChanges();
       }
    }

    /// <summary>
    /// Raises collection change event.
    /// </summary>
    public void NotifyChanges()
    {
        this.ResumeCollectionChangeNotification();
        var arg
             = new NotifyCollectionChangedEventArgs
                  (NotifyCollectionChangedAction.Reset);
        this.OnCollectionChanged(arg);
    }

    /// <summary>
    /// This method removes the given generic list of items as a range
    /// into current collection by casting them as type T.
    /// It then notifies once after all items are removed.
    /// </summary>
    /// <param name="items">The source collection.</param>
    public void RemoveItems(IList<T> items)
    {
        lock(locker)
        {
           this.SuspendCollectionChangeNotification();
           foreach (var i in items)
           {
             Remove(i);
           }
           this.NotifyChanges();
        }
    }

    /// <summary>
    /// Resumes collection changed notification.
    /// </summary>
    public void ResumeCollectionChangeNotification()
    {
        this.suspendCollectionChangeNotification = false;
    }

    /// <summary>
    /// Suspends collection changed notification.
    /// </summary>
    public void SuspendCollectionChangeNotification()
    {
        this.suspendCollectionChangeNotification = true;
    }

    /// <summary>
    /// This collection changed event performs thread safe event raising.
    /// </summary>
    /// <param name="e">The event argument.</param>
    protected override void OnCollectionChanged(NotifyCollectionChangedEventArgs e)
    {
        // Recommended is to avoid reentry 
        // in collection changed event while collection
        // is getting changed on other thread.
        using (BlockReentrancy())
        {
            if (!this.suspendCollectionChangeNotification)
            {
                NotifyCollectionChangedEventHandler eventHandler = 
                      this.CollectionChanged;
                if (eventHandler == null)
                {
                    return;
                }

                // Walk thru invocation list.
                Delegate[] delegates = eventHandler.GetInvocationList();

                foreach
                (NotifyCollectionChangedEventHandler handler in delegates)
                {
                    // If the subscriber is a DispatcherObject and different thread.
                    DispatcherObject dispatcherObject
                         = handler.Target as DispatcherObject;

                    if (dispatcherObject != null
                           && !dispatcherObject.CheckAccess())
                    {
                        // Invoke handler in the target dispatcher's thread... 
                        // asynchronously for better responsiveness.
                        dispatcherObject.Dispatcher.BeginInvoke
                              (DispatcherPriority.DataBind, handler, this, e);
                    }
                    else
                    {
                        // Execute handler as is.
                        handler(this, e);
                    }
                }
            }
        }
    }
}

此外，位于 ObservableCollection 之上的 ICollectionView 会主动感知更改，并与任何其他源列表相比相对较快地执行过滤、分组和排序。

同样，可观察集合可能不是更快数据更新的完美答案，但它们的工作做得很好。

Answer 2

这是我制作的一些解决方案的汇编。收集的想法改变了第一个答案中的调用。

似乎“重置”操作应该与主线程同步，否则 CollectionView 和 CollectionViewSource 会发生奇怪的事情。

我认为这是因为在“重置”处理程序中尝试立即读取集合内容，并且它们应该已经就位。
如果您执行“重置”异步操作，然后立即添加一些异步项目，则新添加的项目可能会添加两次。

public interface IObservableList<T> : IList<T>, INotifyCollectionChanged
{
}

public class ObservableList<T> : IObservableList<T>
{
    private IList<T> collection = new List<T>();
    public event NotifyCollectionChangedEventHandler CollectionChanged;
    private ReaderWriterLock sync = new ReaderWriterLock();

    protected virtual void OnCollectionChanged(NotifyCollectionChangedEventArgs args)
    {
        if (CollectionChanged == null)
            return;
        foreach (NotifyCollectionChangedEventHandler handler in CollectionChanged.GetInvocationList())
        {
            // If the subscriber is a DispatcherObject and different thread.
            var dispatcherObject = handler.Target as DispatcherObject;

            if (dispatcherObject != null && !dispatcherObject.CheckAccess())
            {
                if ( args.Action == NotifyCollectionChangedAction.Reset )
                    dispatcherObject.Dispatcher.Invoke
                          (DispatcherPriority.DataBind, handler, this, args);
                else
                    // Invoke handler in the target dispatcher's thread... 
                    // asynchronously for better responsiveness.
                    dispatcherObject.Dispatcher.BeginInvoke
                          (DispatcherPriority.DataBind, handler, this, args);
            }
            else
            {
                // Execute handler as is.
                handler(this, args);
            }
        }
    }

    public ObservableList()
    {
    }

    public void Add(T item)
    {
        sync.AcquireWriterLock(Timeout.Infinite);
        try
        {
            collection.Add(item);
            OnCollectionChanged(
                    new NotifyCollectionChangedEventArgs(
                      NotifyCollectionChangedAction.Add, item));
        }
        finally
        {
            sync.ReleaseWriterLock();
        }
    }

    public void Clear()
    {
        sync.AcquireWriterLock(Timeout.Infinite);
        try
        {
            collection.Clear();
            OnCollectionChanged(
                    new NotifyCollectionChangedEventArgs(
                        NotifyCollectionChangedAction.Reset));
        }
        finally
        {
            sync.ReleaseWriterLock();
        }
    }

    public bool Contains(T item)
    {
        sync.AcquireReaderLock(Timeout.Infinite);
        try
        {
            var result = collection.Contains(item);
            return result;
        }
        finally
        {
            sync.ReleaseReaderLock();
        }
    }

    public void CopyTo(T[] array, int arrayIndex)
    {
        sync.AcquireWriterLock(Timeout.Infinite);
        try
        {
            collection.CopyTo(array, arrayIndex);
        }
        finally
        {
            sync.ReleaseWriterLock();
        }
    }

    public int Count
    {
        get
        {
            sync.AcquireReaderLock(Timeout.Infinite);
            try
            {
                return collection.Count;
            }
            finally
            {
                sync.ReleaseReaderLock();
            }
        }
    }

    public bool IsReadOnly
    {
        get { return collection.IsReadOnly; }
    }

    public bool Remove(T item)
    {
        sync.AcquireWriterLock(Timeout.Infinite);
        try
        {
            var index = collection.IndexOf(item);
            if (index == -1)
                return false;
            var result = collection.Remove(item);
            if (result)
                OnCollectionChanged(new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Remove, item, index));
            return result;
        }
        finally
        {
            sync.ReleaseWriterLock();
        }
    }

    public IEnumerator<T> GetEnumerator()
    {
        return collection.GetEnumerator();
    }

    System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
    {
        return collection.GetEnumerator();
    }

    public int IndexOf(T item)
    {
        sync.AcquireReaderLock(Timeout.Infinite);
        try
        {
            var result = collection.IndexOf(item);
            return result;
        }
        finally
        {
            sync.ReleaseReaderLock();
        }
    }

    public void Insert(int index, T item)
    {
        sync.AcquireWriterLock(Timeout.Infinite);
        try
        {
            collection.Insert(index, item);
            OnCollectionChanged(new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Add, item, index));
        }
        finally
        {
            sync.ReleaseWriterLock();
        }
    }

    public void RemoveAt(int index)
    {
        sync.AcquireWriterLock(Timeout.Infinite);
        try
        {
            if (collection.Count == 0 || collection.Count <= index)
                return;
            var item = collection[index];
            collection.RemoveAt(index);
            OnCollectionChanged(
                    new NotifyCollectionChangedEventArgs(
                       NotifyCollectionChangedAction.Remove, item, index));
        }
        finally
        {
            sync.ReleaseWriterLock();
        }
    }

    public T this[int index]
    {
        get
        {
            sync.AcquireReaderLock(Timeout.Infinite);
            try
            {
                var result = collection[index];
                return result;
            }
            finally
            {
                sync.ReleaseReaderLock();
            }
        }
        set
        {
            sync.AcquireWriterLock(Timeout.Infinite);
            try
            {
                if (collection.Count == 0 || collection.Count <= index)
                    return;
                var item = collection[index];
                collection[index] = value;
                OnCollectionChanged(
                        new NotifyCollectionChangedEventArgs(
                           NotifyCollectionChangedAction.Replace, value, item, index));
            }
            finally
            {
                sync.ReleaseWriterLock();
            }
        }

    }
}

Answer 3

我无法添加评论，因为我还不够酷，但分享我遇到的这个问题可能值得发布，即使它不是真正的答案。由于 BeginInvoke，我使用此 FastObservableCollection 不断收到“索引超出范围”异常。显然，通知的更改可以在调用处理程序之前撤消，因此为了解决此问题，我传递了以下内容作为从 OnCollectionChanged 方法调用的 BeginInvoke 的第四个参数（而不是使用事件参数之一）：

dispatcherObject.Dispatcher.BeginInvoke
                          (DispatcherPriority.DataBind, handler, this, new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Reset));

而不是这个：

dispatcherObject.Dispatcher.BeginInvoke
                          (DispatcherPriority.DataBind, handler, this, e);

此修复我遇到的“索引超出范围”问题。这是更详细的解释/代码 snpipet：我在哪里获得线程-安全的 CollectionView？

Answer 4

创建同步 Observable 列表的示例：

newSeries = new XYChart.Series<>();
ObservableList<XYChart.Data<Number, Number>> listaSerie;
listaSerie = FXCollections.synchronizedObservableList(FXCollections.observableList(new ArrayList<XYChart.Data<Number, Number>>()));
newSeries.setData(listaSerie);