我在这里how to create no-duplicates ConcurrentQueue?要求任何ConcurentSet / ConcurentQueue实现,但是没有人提出任何建议。
因此,我决定自己编写,对MSDN示例进行一些修改。
我创建了ConcurentSet
类。我将相同的元素插入两次,并期望仅插入一次,因为在内部使用了Set
:
numbers.Add(i);
Console.WriteLine("Add:{0} Count={1}", i, numbers.Count);
numbers.Add(i);
Console.WriteLine("Add:{0} Count={1}", i, numbers.Count);
但是根据输出元素插入两次:
Add:0 Count=0
Add:0 Count=1
Add:1 Count=2
Add:1 Count=3
Add:2 Count=4
Take:0
Add:2 Count=5
Add:3 Count=6
Add:3 Count=7
Add:4 Count=7
Add:4 Count=8
Take:3
Add:5 Count=9
Add:5 Count=10
Add:6 Count=11
Add:6 Count=12
问题是-为什么我的ConcurentSet实现不能按预期工作?
我期望这样的输出:
Add:0 Count=0
Add:0 Count=0
Add:1 Count=1
Add:1 Count=1
Add:2 Count=2
Take:0
Add:2 Count=1
Add:3 Count=2
Add:3 Count=2
Add:4 Count=3
Add:4 Count=3
Take:3
Add:5 Count=3
Add:5 Count=3
.....
完整清单如下:
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
namespace BCBlockingAccess
{
using System;
using System.Collections.Concurrent;
public class ConcurentSet
: IProducerConsumerCollection<int>
{
private readonly object m_lockObject = new object();
private readonly HashSet<int> m_set = new HashSet<int>();
public void CopyTo(Array array, int index)
{
throw new NotImplementedException();
}
public int Count { get { return m_set.Count; } }
public object SyncRoot { get { return m_lockObject; } }
public bool IsSynchronized { get { return true; } }
public void CopyTo(int[] array, int index)
{
throw new NotImplementedException();
}
public bool TryAdd(int item)
{
lock (m_lockObject)
{
m_set.Add(item);
}
return true;
}
public bool TryTake(out int item)
{
lock (m_lockObject)
{
foreach (var i in m_set)
{
if (m_set.Remove(i))
{
item = i;
return true;
}
}
item = -1;
return false;
}
}
public int[] ToArray()
{
throw new NotImplementedException();
}
public IEnumerator<int> GetEnumerator()
{
throw new NotImplementedException();
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
}
class Program
{
static void Main(string[] args)
{
// Increase or decrease this value as desired.
int itemsToAdd = 50;
// Preserve all the display output for Adds and Takes
Console.SetBufferSize(80, (itemsToAdd * 5) + 3);
// A bounded collection. Increase, decrease, or remove the
// maximum capacity argument to see how it impacts behavior.
BlockingCollection<int> numbers = new BlockingCollection<int>(new ConcurentSet());
// A simple blocking consumer with no cancellation.
Task.Factory.StartNew(() =>
{
int i = -1;
while (!numbers.IsCompleted)
{
try
{
i = numbers.Take();
}
catch (InvalidOperationException)
{
Console.WriteLine("Adding was compeleted!");
break;
}
Console.WriteLine("Take:{0} ", i);
// Simulate a slow consumer. This will cause
// collection to fill up fast and thus Adds wil block.
Thread.SpinWait(100000);
}
Console.WriteLine("\r\nNo more items to take. Press the Enter key to exit.");
});
// A simple blocking producer with no cancellation.
Task.Factory.StartNew(() =>
{
for (int i = 0; i < itemsToAdd; i++)
{
numbers.Add(i);
Console.WriteLine("Add:{0} Count={1}", i, numbers.Count);
numbers.Add(i);
Console.WriteLine("Add:{0} Count={1}", i, numbers.Count);
}
// See documentation for this method.
numbers.CompleteAdding();
});
// Keep the console display open in debug mode.
Console.ReadLine();
}
}
}
最佳答案
BlockingCollection保持自己的计数,并且不使用您的基础ConcurrentSet的计数,因此,即使忽略重复项,计数也会增加。
您可能希望围绕此集合编写自己的包装器,该包装器从并发集返回计数,并将所有其他方法属性中继到阻塞集合。