6

BlockingCollection でラップして ConcurrentDictionary を実装しようとしましたが、成功していないようです。

ConcurrentBag<T>、ConcurrentQueue<T>などの BlockingCollection で 1 つの変数宣言が機能することを理解しています。

したがって、BlockingCollection にラップされた ConcurrentBag を作成するには、次のように宣言してインスタンス化します。

BlockingCollection<int> bag = new BlockingCollection<int>(new ConcurrentBag<int>());

しかし、ConcurrentDictionary に対してそれを行う方法は? プロデューサ側とコンシューマ側の両方で BlockingCollection のブロッキング機能が必要です。

4

2 に答える 2

4

多分あなたはblockingCollectionの同時辞書が必要です

        ConcurrentDictionary<int, BlockingCollection<string>> mailBoxes = new ConcurrentDictionary<int, BlockingCollection<string>>();
        int maxBoxes = 5;

        CancellationTokenSource cancelationTokenSource = new CancellationTokenSource();
        CancellationToken cancelationToken = cancelationTokenSource.Token;

        Random rnd = new Random();
        // Producer
        Task.Factory.StartNew(() =>
        {
            while (true)
            {
                int index = rnd.Next(0, maxBoxes);
                // put the letter in the mailbox 'index'
                var box = mailBoxes.GetOrAdd(index, new BlockingCollection<string>());
                box.Add("some message " + index, cancelationToken);
                Console.WriteLine("Produced a letter to put in box " + index);

                // Wait simulating a heavy production item.
                Thread.Sleep(1000);
            }
        });

        // Consumer 1
        Task.Factory.StartNew(() =>
        {
            while (true)
            {
                int index = rnd.Next(0, maxBoxes);
                // get the letter in the mailbox 'index'
                var box = mailBoxes.GetOrAdd(index, new BlockingCollection<string>());
                var message = box.Take(cancelationToken);
                Console.WriteLine("Consumed 1: " + message);

                // consume a item cost less than produce it:
                Thread.Sleep(50);
            }
        });

        // Consumer 2
        Task.Factory.StartNew(() =>
        {
            while (true)
            {
                int index = rnd.Next(0, maxBoxes);
                // get the letter in the mailbox 'index'
                var box = mailBoxes.GetOrAdd(index, new BlockingCollection<string>());
                var message = box.Take(cancelationToken);
                Console.WriteLine("Consumed 2: " + message);

                // consume a item cost less than produce it:
                Thread.Sleep(50);
            }
        });

        Console.ReadLine();
        cancelationTokenSource.Cancel();

このようにして、メールボックス 5 に何かを期待している消費者は、生産者がメールボックス 5 に手紙を入れるまで待ちます。

于 2013-01-30T14:48:13.817 に答える
1

次のような独自のアダプタ クラスを作成する必要があります。

    public class ConcurrentDictionaryWrapper<TKey,TValue> : IProducerConsumerCollection<KeyValuePair<TKey,TValue>>
{
    private ConcurrentDictionary<TKey, TValue> dictionary;

    public IEnumerator<KeyValuePair<TKey, TValue>> GetEnumerator()
    {
        return dictionary.GetEnumerator();
    }

    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }

    public void CopyTo(Array array, int index)
    {
        throw new NotImplementedException();
    }

    public int Count
    {
        get { return dictionary.Count; }
    }

    public object SyncRoot
    {
        get { return this; }
    }

    public bool IsSynchronized
    {
        get { return true; }
    }

    public void CopyTo(KeyValuePair<TKey, TValue>[] array, int index)
    {
        throw new NotImplementedException();
    }

    public bool TryAdd(KeyValuePair<TKey, TValue> item)
    {
        return dictionary.TryAdd(item.Key, item.Value);
    }

    public bool TryTake(out KeyValuePair<TKey, TValue> item)
    {
        item = dictionary.FirstOrDefault();
        TValue value;
        return dictionary.TryRemove(item.Key, out value);
    }

    public KeyValuePair<TKey, TValue>[] ToArray()
    {
        throw new NotImplementedException();
    }
}
于 2012-05-24T13:19:06.097 に答える