BlockingCollection + UI Thread

Question

Я следовал этому руководству , чтобы создать очередь с приоритетами и обернуть ее коллекцией блокировок. У меня есть DataGrid, который я подключил к основной очереди приоритетов, которая генерирует события изменения. Я могу добавить элементы в коллекцию из потока пользовательского интерфейса без заминки, и он блокируется, когда буфер заполнен, как и предполагалось.

Теперь, как мне потреблять предметы? Вот что у меня есть:

public DownloadViewModel()
{
    Queue = new ConcurrentPriorityQueue<DownloadItem>(10);
    Buffer = new BlockingCollection<KeyValuePair<int, DownloadItem>>(Queue, 10000);

    Task.Factory.StartNew(() =>
    {
        KeyValuePair<int, DownloadItem> item;
        while(!Buffer.IsCompleted)
        {
            if(Buffer.TryTake(out item))
            {
                // do something with the item
            }

            Thread.SpinWait(100000);
        }
    });
}

Но как только я добавил этот бит Task.Factory.StartNew, моему приложению внезапно требуется 30 секунд, прежде чем появится окно (до того, как оно было мгновенным), и когда я добавляю элемент, я получаю исключение

Этот тип CollectionView не поддерживает изменения в его SourceCollection из потока, отличного от потока Dispatcher.

Что я понимаю, но действительно ли необходимо брать предметы, используя поток пользовательского интерфейса? Разве это не противоречит цели использования BlockingCollection? Я хочу создать 4 или 8 потребителей и заставить их работать параллельно.

Как это должно быть сделано?

Answer 1

Завершение события CollectionChanged с диспетчером, кажется, работает очень хорошо ...

public bool TryAdd(KeyValuePair<int, T> item)
{
    int pos = _queues.Take(item.Key + 1).Sum(q => q.Count);
    _queues[item.Key].Enqueue(item.Value);
    Interlocked.Increment(ref _count);
    Dispatcher.BeginInvoke(
        new Action(
            () =>
            NotifyCollectionChanged(
                new NotifyCollectionChangedEventArgs(NotifyCollectionChangedAction.Add, item, pos))
        ));
    return true;
}

Просто нужно было получить мой ConcurrentPriorityQueue из DispatcherObject.Я думаю вот как это должно быть сделано.

---

Еще проще, просто напишите метод NotifyCollectionChanged следующим образом:

private void NotifyCollectionChanged(NotifyCollectionChangedEventArgs e)
{
    lock (CollectionChanged)
    {
        if (CollectionChanged != null)
            Dispatcher.BeginInvoke(new Action(() => CollectionChanged(this, e)));
    }
}

А потомвам не нужно мусорить другими вашими методами BeginInvoke.

Answer 2

[После комментирования вопроса, затем]

Вам не нужно "брать предметы, используя поток пользовательского интерфейса". Однако любые обновления пользовательского интерфейса в результате обработки элемента в потребляющей задаче должны быть направлены в поток пользовательского интерфейса. Раздели свои заботы!