数据绑定到 BlockingCollection

Question

我正在尝试实现以下要求（C# 4.0）；

• 一个提交上传请求的“生产者”（在 UI 线程上 - 由用户操作驱动）
• ListView 控件，数据绑定到待处理上传请求的集合
• 处理请求 (FIFO) 并在项目状态更新或请求已完成时保持 UI 最新的消费者。

到目前为止，我还没有弄清楚如何在不使用两个集合的情况下完成上述操作，如下所示；

public void AddUploadRequest(UploadRequest uploadRequest)
{
    this.UploadRequests.Add(uploadRequest);
    this.uploadRequestsBlocking.Add(uploadRequest);
}

...其中UploadRequests 是ObservableCollection<UploadRequest>，uploadRequestsBlocking 是BlockingCollection<UploadRequest>

完整的代码可以在这里找到； http://pastebin.com/620EqaY5（忽略 Dispatcher 的总注入 - 目前只是原型代码）

我通知我的 UI 更新如下；

this.dispatcher.Invoke(() => this.UploadRequests.Remove(uploadRequest));

有没有更好的方法来实现这个功能？或者，更重要的是，这种方法有什么严重的缺陷吗？

我正在考虑的一个可能的扩展是拥有多个消费者，通过使用Parallel.ForEach 而不是GetConsumingPartitioner（基于this 示例）。目前的方法是否会使这不合适？它确实工作正常，但我不能 100% 确信我在此过程中没有在某个地方犯下一些主要的线程错误。

Answer 1

您不能只将数据绑定到 BlockingCollection 的主要（唯一）原因是它没有实现 INotifyCollectionChanged，因此当添加/删除请求时 UI 不会更新。也就是说，有BlockingCollection wrappers around on the net 实现了 INotifyCollectionChanged。或者，您可以改为为 BlockingCollection 公开一个 CollectionView，而不是从 ObservableCollection 调度删除，只需调度 CollectionView 的 Refresh。

    public ICollectionView UploadRequestsView {get;set;}

    public UploadRequester(Dispatcher dispatcher)
    {
        this.dispatcher = dispatcher;           
        this.uploadRequestsBlocking = new BlockingCollection<UploadRequest>();

        UploadRequestsView = CollectionViewSource.GetDefaultView(uploadRequestsBlocking);

        this.consumerTask = Task.Factory.StartNew(this.ConsumeUploadRequests);
    }

    public void AddUploadRequest(UploadRequest uploadRequest)
    {
        uploadRequestsBlocking.Add(uploadRequest);
        UploadRequestsView.Refresh()
    }

    private void ConsumeUploadRequests()
    {
        foreach (var uploadRequest in this.uploadRequestsBlocking.GetConsumingEnumerable())
        {
            uploadRequest.Status = "Uploading...";

            Thread.Sleep(2000);
            uploadRequest.Status = "Successfully uploaded";

            Thread.Sleep(500);
            dispatcher.Invoke(() => UploadRequestsView.Refresh());
        }
    }

Answer 2

我会使用TPL，并且不需要注入Dispatcher，因为TaskScheduler.FromCurrentSynchronizationContext()会给我MainUI线程的当前Dispatcher。

public class UploadRequestProcessor
{
        public ObservableCollection<UploadRequest> UploadRequests { get; private set; }
        private readonly BlockingCollection<UploadRequest> uploadRequestsBlocking;
        private Task consumerTask;

        public UploadRequester()
        {
                this.UploadRequests = new ObservableCollection<UploadRequest>();
                this.uploadRequestsBlocking = new BlockingCollection<UploadRequest>();
                this.consumerTask = Task.Factory.StartNew(this.ConsumeUploadRequests).ContinueWith(nextTask => RemoveUploadRequests(nextTask.Result as BlockingCollection<UploadRequest>), TaskScheduler.FromCurrentSynchronizationContext());
        }

        public void AddUploadRequest(UploadRequest uploadRequest)
        {
                this.UploadRequests.Add(uploadRequest);
                this.uploadRequestsBlocking.Add(uploadRequest);
        }

        private void ConsumeUploadRequests()
        {
            var requestsToRemove = new BlockingCollection<UploadRequest>();
                foreach (var uploadRequest in this.uploadRequestsBlocking.GetConsumingEnumerable())
                {
                        uploadRequest.Status = "Uploading...";

                        Thread.Sleep(2000);
                        uploadRequest.Status = "Successfully uploaded";

                        Thread.Sleep(500);
                        requestsToRemove.Add(uploadRequest);

                }
            return requestsToRemove;
        }

        private void RemoveUploadRequests(BlockingCollection<UploadRequest> requestsToRemove)
        {
            foreach(var request in requestsToRemove)
            {
                UploadRequests.Remove(request);
            }
        }
}