방법: ForEach를 사용하여 BlockingCollection의 항목 제거
Take 및 TryTake 메서드를 사용하여 BlockingCollection<T>에서 항목을 가져오는 것 외에도 foreach(Visual Basic의 경우 For Each)를 사용하여 추가 작업이 완료되고 컬렉션이 빌 때까지 항목을 제거할 수 있습니다. 이를 열거형 변경 또는 열거형 소비라고 합니다. 일반적인 foreach(For Each) 루프와 달리 이 열거자는 항목을 제거하여 소스 컬렉션을 수정하기 때문입니다.
예제
다음 예제에서는 foreach(For Each) 루프를 사용하여 BlockingCollection<T>의 항목을 모두 제거하는 방법을 보여 줍니다.
Option Strict On
Option Explicit On
Imports System.Diagnostics
Imports System.Threading
Imports System.Threading.Tasks
Imports System.Collections.Concurrent
Module EnumerateBC
Class Program
' Limit the collection size to 2000 items
' at any given time. Set itemsToProduce to >500
' to hit the limit.
Const upperLimit As Integer = 1000
' Adjust this number to see how it impacts
' the producing-consuming pattern.
Const itemsToProduce As Integer = 100
Shared collection As BlockingCollection(Of Long) = New BlockingCollection(Of Long)(upperLimit)
' Variables for diagnostic output only.
Shared sw As New Stopwatch()
Shared totalAdditions As Integer = 0
' Counter for synchronizing producers.
Shared producersStillRunning As Integer = 2
Shared Sub Main()
' Start the stopwatch.
sw.Start()
' Queue the Producer threads.
Dim task1 = Task.Factory.StartNew(Sub() RunProducer("A", 0))
Dim task2 = Task.Factory.StartNew(Sub() RunProducer("B", itemsToProduce))
' Store in an array for use with ContinueWhenAll
Dim producers() As Task = {task1, task2}
' Create a cleanup task that will call CompleteAdding after
' all producers are done adding items.
Dim cleanup As Task = Task.Factory.ContinueWhenAll(producers, Sub(p) collection.CompleteAdding())
' Queue the Consumer thread. Put this call
' before Parallel.Invoke to begin consuming as soon as
' the producers add items.
Task.Factory.StartNew(Sub() RunConsumer())
' Keep the console window open while the
' consumer thread completes its output.
Console.ReadKey()
End Sub
Shared Sub RunProducer(ByVal ID As String, ByVal start As Integer)
Dim additions As Integer = 0
For i As Integer = start To start + itemsToProduce - 1
' The data that is added to the collection.
Dim ticks As Long = sw.ElapsedTicks
'Display additions and subtractions.
Console.WriteLine("{0} adding tick value {1}. item# {2}", ID, ticks, i)
' Don't try to add item after CompleteAdding
' has been called.
If collection.IsAddingCompleted = False Then
collection.Add(ticks)
End If
' Counter for demonstration purposes only.
additions = additions + 1
' Uncomment this line to
' slow down the producer threads without sleeping.
Thread.SpinWait(100000)
Next
Interlocked.Add(totalAdditions, additions)
Console.WriteLine("{0} is done adding: {1} items", ID, additions)
End Sub
Shared Sub RunConsumer()
' GetConsumingEnumerable returns the enumerator for the
' underlying collection.
Dim subtractions As Integer = 0
For Each item In collection.GetConsumingEnumerable
subtractions = subtractions + 1
Console.WriteLine("Consuming tick value {0} : item# {1} : current count = {2}",
item.ToString("D18"), subtractions, collection.Count)
Next
Console.WriteLine("Total added: {0} Total consumed: {1} Current count: {2} ",
totalAdditions, subtractions, collection.Count())
sw.Stop()
Console.WriteLine("Press any key to exit.")
End Sub
End Class
End Module
namespace EnumerateBlockingCollection
{
using System;
using System.Collections.Concurrent;
using System.Diagnostics;
class Program
{
// Limit the collection size to 2000 items
// at any given time. Set itemsToProduce to >500
// to hit the limit.
const int upperLimit = 1000;
// Adjust this number to see how it impacts
// the producing-consuming pattern.
const int itemsToProduce = 100;
static BlockingCollection<long> collection = new BlockingCollection<long>(upperLimit);
// Variables for diagnostic output only.
static Stopwatch sw = new Stopwatch();
static int totalAdditions = 0;
// Counter for synchronizing producers.
static int producersStillRunning = 2;
static void Main(string[] args)
{
// Start the stopwatch.
sw.Start();
// Queue the Producer threads. Store in an array
// for use with ContinueWhenAll
Task[] producers = new Task[2];
producers[0] = Task.Factory.StartNew(() => RunProducer("A", 0));
producers[1] = Task.Factory.StartNew(() => RunProducer("B", itemsToProduce));
// Create a cleanup task that will call CompleteAdding after
// all producers are done adding items.
Task cleanup = Task.Factory.ContinueWhenAll(producers, (p) => collection.CompleteAdding());
// Queue the Consumer thread. Put this call
// before Parallel.Invoke to begin consuming as soon as
// the producers add items.
Task.Factory.StartNew(() => RunConsumer());
// Keep the console window open while the
// consumer thread completes its output.
Console.ReadKey();
}
static void RunProducer(string ID, int start)
{
int additions = 0;
for (int i = start; i < start + itemsToProduce; i++)
{
// The data that is added to the collection.
long ticks = sw.ElapsedTicks;
// Display additions and subtractions.
Console.WriteLine("{0} adding tick value {1}. item# {2}", ID, ticks, i);
if(!collection.IsAddingCompleted)
collection.Add(ticks);
// Counter for demonstration purposes only.
additions++;
// Uncomment this line to
// slow down the producer threads ing.
Thread.SpinWait(100000);
}
Interlocked.Add(ref totalAdditions, additions);
Console.WriteLine("{0} is done adding: {1} items", ID, additions);
}
static void RunConsumer()
{
// GetConsumingEnumerable returns the enumerator for the
// underlying collection.
int subtractions = 0;
foreach (var item in collection.GetConsumingEnumerable())
{
Console.WriteLine("Consuming tick value {0} : item# {1} : current count = {2}",
item.ToString("D18"), subtractions++, collection.Count);
}
Console.WriteLine("Total added: {0} Total consumed: {1} Current count: {2} ",
totalAdditions, subtractions, collection.Count());
sw.Stop();
Console.WriteLine("Press any key to exit");
}
}
}
이 예제에서는 foreach 루프를 소비 스레드의 BlockingCollection<T>.GetConsumingEnumerable 메서드와 함께 사용하여 컬렉션을 열거할 때 해당 컬렉션에서 각 항목을 제거합니다. System.Collections.Concurrent.BlockingCollection<T>은 컬렉션에 있는 항목의 최대 수를 항상 제한합니다. 이와 같이 컬렉션을 열거하는 경우 사용 가능한 항목이 없거나 컬렉션이 비어 있으면 소비자 스레드가 차단됩니다. 이 예제에서는 생산자 스레드가 항목이 소비되는 것보다 더 빠르게 항목을 추가하기 때문에 차단이 발생하지 않습니다.
생산자 스레드를 통해 항목이 추가된 순서와 항목의 열거 순서는 같지 않을 수 있습니다.
컬렉션을 수정하지 않고 열거하려면 GetConsumingEnumerable 메서드 없이 foreach(For Each)만 사용합니다. 그러나 이와 같이 열거한 결과는 특정 시점에서의 컬렉션을 나타내는 스냅숏에 불과하다는 사실을 이해하고 있어야 합니다. 루프를 실행하는 동안 다른 스레드를 통해 항목이 계속하여 추가 또는 제거되고 있다면 루프에 표시되는 상태가 컬렉션의 실제 상태와 다를 수 있습니다.