Usuwanie elementów w kolekcji BlockingCollection za pomocą foreach
Oprócz pobierania elementów z BlockingCollection<T> obiektu przy użyciu Take metody i TryTake można również użyć foreach (dla każdego w Visual Basic) z elementem BlockingCollection<T>.GetConsumingEnumerable , aby usunąć elementy do momentu ukończenia dodawania, a kolekcja jest pusta. Jest to nazywane wyliczeniemmutującymlub zużywającym wyliczenie, ponieważ w przeciwieństwie do typowej foreach pętli (For Each) ten moduł wyliczający modyfikuje kolekcję źródłową, usuwając elementy.
Przykład
W poniższym przykładzie pokazano, jak usunąć wszystkie elementy w BlockingCollection<T> pętli (For Each).foreach
using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
class Example
{
// 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 readonly BlockingCollection<long> Collection =
new BlockingCollection<long>(UpperLimit);
// Variables for diagnostic output only.
static readonly Stopwatch Stopwatch = new Stopwatch();
static int TotalAdditions = 0;
static async Task Main()
{
Stopwatch.Start();
// Queue the consumer task.
var consumerTask = Task.Run(() => RunConsumer());
// Queue the producer tasks.
var produceTaskOne = Task.Run(() => RunProducer("A", 0));
var produceTaskTwo = Task.Run(() => RunProducer("B", ItemsToProduce));
var producerTasks = new[] { produceTaskOne , produceTaskTwo };
// Create a cleanup task that will call CompleteAdding after
// all producers are done adding items.
var cleanupTask = Task.Factory.ContinueWhenAll(producerTasks, _ => Collection.CompleteAdding());
// Wait for all tasks to complete
await Task.WhenAll(consumerTask, produceTaskOne, produceTaskTwo, cleanupTask);
// Keep the console window open while the
// consumer thread completes its output.
Console.WriteLine("Press any key to exit");
Console.ReadKey(true);
}
static void RunProducer(string id, int start)
{
var additions = 0;
for (var i = start; i < start + ItemsToProduce; i++)
{
// The data that is added to the collection.
var ticks = Stopwatch.ElapsedTicks;
// Display additions and subtractions.
Console.WriteLine($"{id} adding tick value {ticks}. item# {i}");
if (!Collection.IsAddingCompleted)
{
Collection.Add(ticks);
}
// Counter for demonstration purposes only.
additions++;
// Comment this line to speed up the producer threads.
Thread.SpinWait(100000);
}
Interlocked.Add(ref TotalAdditions, additions);
Console.WriteLine($"{id} is done adding: {additions} items");
}
static void RunConsumer()
{
// GetConsumingEnumerable returns the enumerator for the underlying collection.
var subtractions = 0;
foreach (var item in Collection.GetConsumingEnumerable())
{
Console.WriteLine(
$"Consuming tick value {item:D18} : item# {subtractions++} : current count = {Collection.Count}");
}
Console.WriteLine(
$"Total added: {TotalAdditions} Total consumed: {subtractions} Current count: {Collection.Count}");
Stopwatch.Stop();
}
}
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
W tym przykładzie użyto foreach pętli z BlockingCollection<T>.GetConsumingEnumerable metodą w wątku zużywającym, co powoduje usunięcie każdego elementu z kolekcji podczas wyliczania. System.Collections.Concurrent.BlockingCollection<T> ogranicza maksymalną liczbę elementów znajdujących się w kolekcji w dowolnym momencie. Wyliczanie kolekcji w ten sposób blokuje wątek odbiorcy, jeśli nie są dostępne elementy lub czy kolekcja jest pusta. W tym przykładzie blokowanie nie jest problemem, ponieważ wątek producenta dodaje elementy szybciej niż mogą być używane.
Zwraca BlockingCollection<T>.GetConsumingEnumerable wartość IEnumerable<T>, dlatego nie można zagwarantować kolejności. Jednak wewnętrznie System.Collections.Concurrent.ConcurrentQueue<T> element jest używany jako typ kolekcji bazowej , który spowoduje oddzielenie obiektów od kolejki po porządkoweniu pierwszego na pierwszym wyjęcie (FIFO). Jeśli wykonywane są współbieżne wywołania BlockingCollection<T>.GetConsumingEnumerable , będą konkurować. Nie można zaobserwować jednego elementu zużytego (wyczekiwanego) w jednym wyliczenie w drugim.
Aby wyliczyć kolekcję bez jej modyfikowania, wystarczy użyć foreach metody (For Each) bez GetConsumingEnumerable metody . Należy jednak pamiętać, że tego rodzaju wyliczenie reprezentuje migawkę kolekcji w określonym momencie w czasie. Jeśli inne wątki dodają lub usuwają elementy jednocześnie podczas wykonywania pętli, pętla może nie reprezentować rzeczywistego stanu kolekcji.
