Como: Adicionar a funcionalidade de delimitação e bloqueio a uma coleção
Este exemplo mostra como adicionar a funcionalidade de delimitação e de bloqueio a uma classe de coleção personalizada por meio da implementação da interface System.Collections.Concurrent.IProducerConsumerCollection<T> na classe e do uso de uma instância da classe como o mecanismo de armazenamento interno para um System.Collections.Concurrent.BlockingCollection<T>. Para obter mais informações sobre delimitação e bloqueio, veja Visão geral de BlockingCollection.
Exemplo
A classe de coleção personalizada é uma fila de prioridade básica na qual os níveis de prioridade são representados como uma matriz de objetos System.Collections.Concurrent.ConcurrentQueue<T>. Nenhuma solicitação adicional é executada dentro de cada fila.
No código do cliente, três tarefas são iniciadas. A primeira tarefa apenas monitora a digitação no teclado para permitir o cancelamento a qualquer momento durante a execução. A segunda tarefa é o thread produtor; ele adiciona novos itens na coleção de bloqueio e atribui uma prioridade com base em um valor aleatório a cada item. A terceira tarefa remove itens da coleção assim que eles ficarem disponíveis.
Você pode ajustar o comportamento do aplicativo fazendo com que um dos threads seja executado mais rapidamente do que o outro. Se o produtor for executado mais rapidamente, você observará que a funcionalidade delimitadora como a coleta de bloqueio impedirá que itens sejam adicionados se ela já contiver o número de itens especificado no construtor. Se o consumidor for executado mais rapidamente, você observará a funcionalidade de bloqueio enquanto o consumidor aguarda que um novo item seja adicionado.
namespace ProdConsumerCS
{
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
// Implementation of a priority queue that has bounding and blocking functionality.
public class SimplePriorityQueue<TPriority, TValue> : IProducerConsumerCollection<KeyValuePair<int, TValue>>
{
// Each internal queue in the array represents a priority level.
// All elements in a given array share the same priority.
private ConcurrentQueue<KeyValuePair<int, TValue>>[] _queues = null;
// The number of queues we store internally.
private int priorityCount = 0;
private int m_count = 0;
public SimplePriorityQueue(int priCount)
{
this.priorityCount = priCount;
_queues = new ConcurrentQueue<KeyValuePair<int, TValue>>[priorityCount];
for (int i = 0; i < priorityCount; i++)
_queues[i] = new ConcurrentQueue<KeyValuePair<int, TValue>>();
}
// IProducerConsumerCollection members
public bool TryAdd(KeyValuePair<int, TValue> item)
{
_queues[item.Key].Enqueue(item);
Interlocked.Increment(ref m_count);
return true;
}
public bool TryTake(out KeyValuePair<int, TValue> item)
{
bool success = false;
// Loop through the queues in priority order
// looking for an item to dequeue.
for (int i = 0; i < priorityCount; i++)
{
// Lock the internal data so that the Dequeue
// operation and the updating of m_count are atomic.
lock (_queues)
{
success = _queues[i].TryDequeue(out item);
if (success)
{
Interlocked.Decrement(ref m_count);
return true;
}
}
}
// If we get here, we found nothing.
// Assign the out parameter to its default value and return false.
item = new KeyValuePair<int, TValue>(0, default(TValue));
return false;
}
public int Count
{
get { return m_count; }
}
// Required for ICollection
void ICollection.CopyTo(Array array, int index)
{
CopyTo(array as KeyValuePair<int, TValue>[], index);
}
// CopyTo is problematic in a producer-consumer.
// The destination array might be shorter or longer than what
// we get from ToArray due to adds or takes after the destination array was allocated.
// Therefore, all we try to do here is fill up destination with as much
// data as we have without running off the end.
public void CopyTo(KeyValuePair<int, TValue>[] destination, int destStartingIndex)
{
if (destination == null) throw new ArgumentNullException();
if (destStartingIndex < 0) throw new ArgumentOutOfRangeException();
int remaining = destination.Length;
KeyValuePair<int, TValue>[] temp = this.ToArray();
for (int i = 0; i < destination.Length && i < temp.Length; i++)
destination[i] = temp[i];
}
public KeyValuePair<int, TValue>[] ToArray()
{
KeyValuePair<int, TValue>[] result;
lock (_queues)
{
result = new KeyValuePair<int, TValue>[this.Count];
int index = 0;
foreach (var q in _queues)
{
if (q.Count > 0)
{
q.CopyTo(result, index);
index += q.Count;
}
}
return result;
}
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
public IEnumerator<KeyValuePair<int, TValue>> GetEnumerator()
{
for (int i = 0; i < priorityCount; i++)
{
foreach (var item in _queues[i])
yield return item;
}
}
public bool IsSynchronized
{
get
{
throw new NotSupportedException();
}
}
public object SyncRoot
{
get { throw new NotSupportedException(); }
}
}
public class TestBlockingCollection
{
static void Main()
{
int priorityCount = 7;
SimplePriorityQueue<int, int> queue = new SimplePriorityQueue<int, int>(priorityCount);
var bc = new BlockingCollection<KeyValuePair<int, int>>(queue, 50);
CancellationTokenSource cts = new CancellationTokenSource();
Task.Run(() =>
{
if (Console.ReadKey(true).KeyChar == 'c')
cts.Cancel();
});
// Create a Task array so that we can Wait on it
// and catch any exceptions, including user cancellation.
Task[] tasks = new Task[2];
// Create a producer thread. You can change the code to
// make the wait time a bit slower than the consumer
// thread to demonstrate the blocking capability.
tasks[0] = Task.Run(() =>
{
// We randomize the wait time, and use that value
// to determine the priority level (Key) of the item.
Random r = new Random();
int itemsToAdd = 40;
int count = 0;
while (!cts.Token.IsCancellationRequested && itemsToAdd-- > 0)
{
int waitTime = r.Next(2000);
int priority = waitTime % priorityCount;
var item = new KeyValuePair<int, int>(priority, count++);
bc.Add(item);
Console.WriteLine("added pri {0}, data={1}", item.Key, item.Value);
}
Console.WriteLine("Producer is done adding.");
bc.CompleteAdding();
},
cts.Token);
//Give the producer a chance to add some items.
Thread.SpinWait(1000000);
// Create a consumer thread. The wait time is
// a bit slower than the producer thread to demonstrate
// the bounding capability at the high end. Change this value to see
// the consumer run faster to demonstrate the blocking functionality
// at the low end.
tasks[1] = Task.Run(() =>
{
while (!bc.IsCompleted && !cts.Token.IsCancellationRequested)
{
Random r = new Random();
int waitTime = r.Next(2000);
Thread.SpinWait(waitTime * 70);
// KeyValuePair is a value type. Initialize to avoid compile error in if(success)
KeyValuePair<int, int> item = new KeyValuePair<int, int>();
bool success = false;
success = bc.TryTake(out item);
if (success)
{
// Do something useful with the data.
Console.WriteLine("removed Pri = {0} data = {1} collCount= {2}", item.Key, item.Value, bc.Count);
}
else
{
Console.WriteLine("No items to retrieve. count = {0}", bc.Count);
}
}
Console.WriteLine("Exited consumer loop");
},
cts.Token);
try {
Task.WaitAll(tasks, cts.Token);
}
catch (OperationCanceledException e) {
if (e.CancellationToken == cts.Token)
Console.WriteLine("Operation was canceled by user. Press any key to exit");
}
catch (AggregateException ae) {
foreach (var v in ae.InnerExceptions)
Console.WriteLine(v.Message);
}
finally {
cts.Dispose();
}
Console.ReadKey(true);
}
}
}
Por padrão, o armazenamento para um System.Collections.Concurrent.BlockingCollection<T> é System.Collections.Concurrent.ConcurrentQueue<T>.
