BlockingCollection 非常容易做到这一点。
var batchingQueue = new BlockingCollection<BrokeredMessage>();
myQueueClient.OnMessage((m) =>
{
Console.WriteLine("Queueing message");
batchingQueue.Add(m);
});
还有你的消费者线程：
foreach (var msg in batchingQueue.GetConsumingEnumerable())
{
Console.WriteLine("Completing message");
msg.Complete();
}
GetConsumingEnumerable 返回一个迭代器，它消耗队列中的项目，直到设置了IsCompleted 属性并且队列为空。如果队列为空但IsCompleted 为False，则它会进行非忙等待下一项。
要取消消费者线程（即关闭程序），您停止向队列添加内容并让主线程调用batchingQueue.CompleteAdding。消费者将队列清空，看到IsCompleted属性为True，然后退出。
在这里使用BlockingCollection 比ConcurrentBag 或ConcurrentQueue 更好，因为BlockingCollection 接口更易于使用。特别是，GetConsumingEnumerable 的使用使您不必担心检查计数或忙等待（轮询循环）。它只是工作。
还要注意ConcurrentBag 有一些相当奇怪的删除行为。特别是，删除项目的顺序会根据删除项目的线程而有所不同。创建袋子的线程以与其他线程不同的顺序移除项目。详情请见Using the ConcurrentBag Collection。
您还没有说明为什么要在输入时对项目进行批处理。除非有压倒一切的性能原因这样做，否则使用批处理逻辑使代码复杂化似乎不是一个特别好的主意。
如果您想批量写入数据库，那么我建议使用简单的List<T> 来缓冲项目。如果您必须在将项目写入数据库之前对其进行处理，请使用我上面展示的技术来处理它们。然后，与其直接写入数据库，不如将项目添加到列表中。当列表获得 1,000 个项目或经过给定的时间量时，分配一个新列表并启动一个任务以将旧列表写入数据库。像这样：
// at class scope
// Flush every 5 minutes.
private readonly TimeSpan FlushDelay = TimeSpan.FromMinutes(5);
private const int MaxBufferItems = 1000;
// Create a timer for the buffer flush.
System.Threading.Timer _flushTimer = new System.Threading.Timer(TimedFlush, FlushDelay.TotalMilliseconds, Timeout.Infinite);
// A lock for the list. Unless you're getting hundreds of thousands
// of items per second, this will not be a performance problem.
object _listLock = new Object();
List<BrokeredMessage> _recordBuffer = new List<BrokeredMessage>();
然后，在您的消费者中：
foreach (var msg in batchingQueue.GetConsumingEnumerable())
{
// process the message
Console.WriteLine("Completing message");
msg.Complete();
lock (_listLock)
{
_recordBuffer.Add(msg);
if (_recordBuffer.Count >= MaxBufferItems)
{
// Stop the timer
_flushTimer.Change(Timeout.Infinite, Timeout.Infinite);
// Save the old list and allocate a new one
var myList = _recordBuffer;
_recordBuffer = new List<BrokeredMessage>();
// Start a task to write to the database
Task.Factory.StartNew(() => FlushBuffer(myList));
// Restart the timer
_flushTimer.Change(FlushDelay.TotalMilliseconds, Timeout.Infinite);
}
}
}
private void TimedFlush()
{
bool lockTaken = false;
List<BrokeredMessage> myList = null;
try
{
if (Monitor.TryEnter(_listLock, 0, out lockTaken))
{
// Save the old list and allocate a new one
myList = _recordBuffer;
_recordBuffer = new List<BrokeredMessage>();
}
}
finally
{
if (lockTaken)
{
Monitor.Exit(_listLock);
}
}
if (myList != null)
{
FlushBuffer(myList);
}
// Restart the timer
_flushTimer.Change(FlushDelay.TotalMilliseconds, Timeout.Infinite);
}
这里的想法是，您将旧列表移开，分配一个新列表以便继续处理，然后将旧列表的项目写入数据库。锁是为了防止计时器和记录计数器相互踩踏。如果没有锁，事情可能会在一段时间内正常运行，然后您会在不可预知的时间发生奇怪的崩溃。
我喜欢这种设计，因为它消除了消费者的轮询。我唯一不喜欢的是消费者必须知道计时器（即它必须停止然后重新启动计时器）。稍加思考，我就可以消除这个要求。但它的编写方式效果很好。