Неблокирующая параллельная коллекция?

Я думаю, что у меня есть два возможных решения. Я не очень доволен ни тем, ни другим, но они, по крайней мере, представляют собой разумную альтернативу подходу APM.

Первый не соответствует вашему требованию отсутствия блокирующего потока, но я думаю, что он довольно элегантен, потому что вы можете регистрировать обратные вызовы, и они будут вызываться в циклическом режиме, но у вас все еще есть возможность вызывать Take или TryTake, как обычно. для BlockingCollection. Этот код принудительно регистрирует обратные вызовы каждый раз, когда запрашивается элемент. Это сигнальный механизм для коллекции. Преимущество этого подхода в том, что вызовы Take не прерываются, как в моем втором решении.

public class NotifyingBlockingCollection<T> : BlockingCollection<T>
{
    private Thread m_Notifier;
    private BlockingCollection<Action<T>> m_Callbacks = new BlockingCollection<Action<T>>();

    public NotifyingBlockingCollection()
    {
        m_Notifier = new Thread(Notify);
        m_Notifier.IsBackground = true;
        m_Notifier.Start();
    }

    private void Notify()
    {
        while (true)
        {
            Action<T> callback = m_Callbacks.Take();
            T item = Take();
            callback.BeginInvoke(item, null, null); // Transfer to the thread pool.
        }
    }

    public void RegisterForTake(Action<T> callback)
    {
        m_Callbacks.Add(callback);
    }
}

Второй соответствует вашему требованию отсутствия блокирующего потока. Обратите внимание, как он передает вызов обратного вызова в пул потоков. Я сделал это, потому что думаю, что если бы он выполнялся синхронно, то блокировки удерживались бы дольше, что приводило к узким местам Add и RegisterForTake. Я внимательно изучил его и не думаю, что он может быть заблокирован в реальном времени (доступны и элемент, и обратный вызов, но обратный вызов никогда не выполняется), но вы можете просмотреть его самостоятельно, чтобы убедиться. Единственная проблема здесь заключается в том, что вызов Take будет зависать, поскольку обратные вызовы всегда имеют приоритет.

public class NotifyingBlockingCollection<T>
{
    private BlockingCollection<T> m_Items = new BlockingCollection<T>();
    private Queue<Action<T>> m_Callbacks = new Queue<Action<T>>();

    public NotifyingBlockingCollection()
    {
    }

    public void Add(T item)
    {
        lock (m_Callbacks)
        {
            if (m_Callbacks.Count > 0)
            {
                Action<T> callback = m_Callbacks.Dequeue();
                callback.BeginInvoke(item, null, null); // Transfer to the thread pool.
            }
            else
            {
                m_Items.Add(item);
            }
        }
    }

    public T Take()
    {
        return m_Items.Take();
    }

    public void RegisterForTake(Action<T> callback)
    {
        lock (m_Callbacks)
        {
            T item;
            if (m_Items.TryTake(out item))
            {
                callback.BeginInvoke(item, null, null); // Transfer to the thread pool.
            }
            else
            {
                m_Callbacks.Enqueue(callback);
            }
        }
    }
}

Как насчет чего-то подобного? (Название, вероятно, требует некоторой работы. И обратите внимание, что это не проверено.)

public class CallbackCollection<T>
{
    // Sychronization object to prevent race conditions.
    private object _SyncObject = new object();

    // A queue for callbacks that are waiting for items.
    private ConcurrentQueue<Action<T>> _Callbacks = new ConcurrentQueue<Action<T>>();

    // A queue for items that are waiting for callbacks.
    private ConcurrentQueue<T> _Items = new ConcurrentQueue<T>();

    public void Add(T item)
    {
        Action<T> callback;
        lock (_SyncObject)
        {
            // Try to get a callback. If no callback is available,
            // then enqueue the item to wait for the next callback
            // and return.
            if (!_Callbacks.TryDequeue(out callback))
            {
                _Items.Enqueue(item);
                return;
            }
        }

        ExecuteCallback(callback, item);
    }

    public void TakeAndCallback(Action<T> callback)
    {
        T item;
        lock(_SyncObject)
        {
            // Try to get an item. If no item is available, then
            // enqueue the callback to wait for the next item
            // and return.
            if (!_Items.TryDequeue(out item))
            {
                _Callbacks.Enqueue(callback);
                return;
            }
        }
        ExecuteCallback(callback, item);
    }

    private void ExecuteCallback(Action<T> callback, T item)
    {
        // Use a new Task to execute the callback so that we don't
        // execute it on the current thread.
        Task.Factory.StartNew(() => callback.Invoke(item));
    }
}