.Net线程安全集合
线程安全集合
.NET Framework 4 引入了 System.Collections.Concurrent 命名空间，其中包含多个线程安全且可缩放的集合类。 多个线程可以安全高效地从这些集合添加或删除项，而无需在用户代码中进行其他同步。 编写新代码时，只要将多个线程同时写入到集合时，就使用并发集合类。 如果仅从共享集合进行读取，则可使用 System.Collections.Generic 命名空间中的类。 建议不要使用 1.0 集合类，除非需要定位 .NET Framework 1.1 或更低版本运行时。
.NET Framework 1.0 和 2.0 集合中的线程同步
.NET Framework 1.0 中引入的集合位于 System.Collections 命名空间中。 这些集合（包括常用的 ArrayList 和 Hashtable）通过 Synchronized 属性（此属性围绕集合返回线程安全的包装器）提供一些线程安全性。 该包装器通过对每个添加或删除操作锁定整个集合进行工作。 因此，每个尝试访问集合的线程必须等待，直到轮到它获取锁定。 这不可缩放，并且可能导致大型集合的性能显著下降。 此外，这一设计并不能完全防止争用情况的出现。 有关详细信息，请参阅泛型集合中的同步。
.NET Framework 2.0 中引入的集合类位于 System.Collections.Generic 命名空间中。 它们包括 List
建议使用 .NET Framework 4 中的并发集合类，因为它们不仅能够提供 .NET Framework 2.0 集合类的类型安全性，而且能够比 .NET Framework 1.0 集合更高效完整地提供线程安全性。
细粒度锁定和无锁机制
某些并发集合类型使用轻量同步机制，如 SpinLock、SpinWait、SemaphoreSlim 和 CountdownEvent，这些机制是 .NET Framework 4 中的新增功能。 这些同步类型通常在将线程真正置于等待状态之前，会在短时间内使用忙旋转。 预计等待时间非常短时，旋转比等待所消耗的计算资源少得多，因为后者涉及资源消耗量大的内核转换。 对于使用旋转的集合类，这种效率意味着多个线程能够以非常快的速率添加和删除项。 有关旋转与锁定的详细信息，请参阅 SpinLock 和 SpinWait。
ConcurrentQueue
备注
由于并发集合类支持 ICollection，因此该类可提供针对 IsSynchronized 和 SyncRoot 属性的实现，即使这些属性不相关。 IsSynchronized 始终返回 false，而 SyncRoot 始终为 null（在 Visual Basic 中是 Nothing）。
下表列出了 System.Collections.Concurrent 命名空间中的集合类型。
| 类型 | 描述 |
|---|---|
| BlockingCollection |
为实现 IProducerConsumerCollection |
| ConcurrentDictionary |
键值对字典的线程安全实现。 |
| ConcurrentQueue |
FIFO（先进先出）队列的线程安全实现。 |
| ConcurrentStack |
LIFO（后进先出）堆栈的线程安全实现。 |
| ConcurrentBag |
无序元素集合的线程安全实现。 |
| IProducerConsumerCollection |
类型必须实现以在 BlockingCollection 中使用的接口。 |
相关主题
| Title | 描述 |
|---|---|
| BlockingCollection 概述 | 描述 BlockingCollection |
| 如何：在 ConcurrentDictionary 中添加和移除项 | 描述如何从 ConcurrentDictionary |
| 如何：在 BlockingCollection 中逐个添加和取出项 | 描述如何在不使用只读枚举器的情况下，从阻止的集合添加和检索项。 |
| 如何：向集合添加限制和阻塞功能 | 描述如何将任一集合类用作 IProducerConsumerCollection |
| 如何：使用 ForEach 移除 BlockingCollection 中的项 | 介绍了如何使用 foreach（在 Visual Basic 中是 For Each）在锁定集合中删除所有项。 |
| 如何：在管道中使用阻塞集合的数组 | 描述如何同时使用多个阻塞集合来实现一个管道。 |
| 如何：使用 ConcurrentBag 创建目标池 | 演示如何使用并发包在可重用对象（而不是继续创建新对象）的情况下改进性能。 |
参考
System.Collections.Concurrent
建议的内容
-
Interlocked.Increment 方法 (System.Threading)
以原子操作的形式递增指定变量的值并存储结果。
-
Semaphore 类 (System.Threading)
限制可同时访问某一资源或资源池的线程数。
-
ConcurrentBag
类 (System.Collections.Concurrent) 表示对象的线程安全的无序集合。
何时使用线程安全集合
了解何时在 .NET 中使用线程安全集合。 有五种专门为支持多线程添加和删除操作而设计的集合类型。
BlockingCollection 概述
BlockingCollection
-
实现制造者-使用者模式。
-
通过多线程并发添加和获取项。
-
可选最大容量。
-
集合为空或已满时通过插入和移除操作进行阻塞。
-
插入和移除“尝试”操作不发生阻塞，或在指定时间段内发生阻塞。
-
封装实现 IProducerConsumerCollection
的任何集合类型 -
使用取消标记执行取消操作。
-
支持使用
foreach（在 Visual Basic 中，使用For Each）的两种枚举：-
只读枚举。
-
在枚举项时将项移除的枚举。
-
限制和阻塞支持
BlockingCollection
多个线程或任务可同时向集合添加项，如果集合达到其指定最大容量，则制造线程将发生阻塞，直到移除集合中的某个项。 多个使用者可以同时移除项，如果集合变空，则使用线程将发生阻塞，直到制造者添加某个项。 制造线程可调用 CompleteAdding 来指示不再添加项。 使用者将监视 IsCompleted 属性以了解集合何时为空且不再添加项。 下面的示例展示了容量上限为 100 的简单 BlockingCollection。 只要满足某些外部条件为 true，制造者任务就会向集合添加项，然后调用 CompleteAdding。 使用者任务获取项，直到 IsCompleted 属性为 true。
C#
{
while (moreItemsToAdd)
{
Data data = GetData();
// Blocks if numbers.Count == dataItems.BoundedCapacity
dataItems.Add(data);
}
// Let consumer know we are done.
dataItems.CompleteAdding();
});
">// A bounded collection. It can hold no more
// than 100 items at once.
BlockingCollection dataItems = new BlockingCollection(100);
// A simple blocking consumer with no cancellation.
Task.Run(() =>
{
while (!dataItems.IsCompleted)
{
Data data = null;
// Blocks if dataItems.Count == 0.
// IOE means that Take() was called on a completed collection.
// Some other thread can call CompleteAdding after we pass the
// IsCompleted check but before we call Take.
// In this example, we can simply catch the exception since the
// loop will break on the next iteration.
try
{
data = dataItems.Take();
}
catch (InvalidOperationException) { }
if (data != null)
{
Process(data);
}
}
Console.WriteLine("\r\nNo more items to take.");
});
// A simple blocking producer with no cancellation.
Task.Run(() =>
{
while (moreItemsToAdd)
{
Data data = GetData();
// Blocks if numbers.Count == dataItems.BoundedCapacity
dataItems.Add(data);
}
// Let consumer know we are done.
dataItems.CompleteAdding();
});
有关完整的示例，请参阅 如何：在 BlockingCollection 中逐个添加和取出项。
计时阻塞操作
在针对有限集合的计时阻塞 TryAdd 和 TryTake 操作中，此方法将尝试添加或取出某个项。 如果项可用，项会被置于通过引用传入的变量中，然后方法返回 true。 如果在指定的超时期限过后未检索到任何项，方法返回 false。 相应线程可以任意执行一些其他有用的工作，然后再重新尝试访问该集合。 有关计时阻塞访问的示例，请参阅如何：在 BlockingCollection 中逐个添加和取出项中的第二个示例。
取消添加和取出操作
添加和取出操作通常会在一个循环内执行。 可以通过以下方法来取消循环：向 TryAdd 或 TryTake 方法传入 CancellationToken，然后在每次迭代时检查该标记的 IsCancellationRequested 属性的值。 如果值为 true，由你自行决定是否通过清理所有资源并退出循环来响应取消请求。 下面的示例演示获取取消标记和使用该标记的代码的 TryAdd 重载：
C#do
{
// Cancellation causes OCE. We know how to handle it.
try
{
success = bc.TryAdd(itemToAdd, 2, ct);
}
catch (OperationCanceledException)
{
bc.CompleteAdding();
break;
}
//...
} while (moreItems == true);
有关如何添加取消支持的示例，请参阅如何：在 BlockingCollection 中逐个添加和取出项中的第二个示例。
指定集合类型
创建 BlockingCollection
BlockingCollection<string> bc = new BlockingCollection<string>(new ConcurrentBag<string>(), 1000 );
有关详细信息，请参阅如何：向集合添加限制和阻塞功能。
IEnumerable 支持
BlockingCollectionforeach（在 Visual Basic 中为 For Each）删除项直到完成集合，也就是说，集合为空且不再添加项。 有关详细信息，请参阅如何：使用 ForEach 移除 BlockingCollection 中的项。
将多个 BlockingCollection 作为整体使用
在使用者需要同时取出多个集合中的项的情况下，可以创建 BlockingCollection
请参阅
- System.Collections.Concurrent
- 集合和数据结构
- 线程安全集合
何时使用线程安全集合
.NET Framework 4 引入了五种专为支持多线程添加和删除操作而设计的集合类型。 为了实现线程安全，这些类型使用多种高效的锁定和免锁定同步机制。 同步会增加操作的开销。 开销数取决于所用的同步类型、执行的操作类型和其他因素，例如尝试并行访问该集合的线程数。
在某些方案中，同步开销可忽略不计，使多线程类型的执行速度和缩放水平远远超过其受外部锁保护的非线程安全同等类型。 在其他方案中，开销可能会导致线程安全类型的执行速度和缩放水平与该类型外部锁定的非线程安全版本相同，甚至更差。
以下部分提供有关何时使用线程安全集合与其非线程安全同等集合（其读写操作受用户提供的锁定保护）的通用指南。 由于性能可能因多种因素而异，所以本指南并不针对某特定情况且不一定对所有情况都有效。 如果性能非常重要，那么确定要使用的集合类型的最佳方式是基于典型计算机配置和负载衡量性能。 本文档使用以下术语：
纯生成方-使用方方案
任何给定线程要么添加元素，要么删除元素，两种操作不能同时执行。
混合生成方-使用方方案
任何给定线程可同时添加和删除元素。
加速
相对于同一方案中其他类型更快速的算法性能。
可伸缩性
与计算机内核数相称的性能提升。 一种可伸缩的算法，相比两个内核，八个内核上的执行速度更快。
ConcurrentQueue (T) 与 Queue(T)
在纯制造者-使用者方案中，每个元素的处理时间都非常短（几条指令），而相比带有外部锁的 System.Collections.Generic.Queue
处理时间大约为 500 FLOPS（浮点运算）或更长时，该双线程规则不适用于 ConcurrentQueue
在混合制造者-使用者方案中，处理时间非常短时，带外部锁的 Queue
ConcurrentStack 与堆栈
在纯制造者-使用者方案中，当处理时间非常短时，System.Collections.Concurrent.ConcurrentStack
在混合制造者-使用者方案中，对于小型和大型工作负荷，ConcurrentStack
使用 PushRange 和 TryPopRange 可能会大大加快访问速度。
ConcurrentDictionary 与词典
通常，在从多个线程中并行添加和更新键或值的任何方案中，会使用 System.Collections.Concurrent.ConcurrentDictionary
在涉及频繁更新的方案中，可以提高 ConcurrentDictionary
如果仅读取键或值，Dictionary
ConcurrentBag
在纯制造者-使用者方案中，System.Collections.Concurrent.ConcurrentBag
在混合制造者-使用者方案中，对于大型和小型工作负荷，相比其他任何并发集合类型，往往 ConcurrentBag
BlockingCollection
需要限制和阻止语义时，System.Collections.Concurrent.BlockingCollection
请参阅
- System.Collections.Concurrent
- 线程安全集合
- 并行编程
如何：在 BlockingCollection 中逐个添加和取出项
此示例展示了如何以阻止性和非阻止性方式在 BlockingCollection
有关如何枚举 BlockingCollection
示例
第一个示例展示了如何添加和取出项，以便在集合暂时为空（取出时）或达到最大容量（添加时），或超过指定超时期限时，阻止相应操作。 注意，仅当已创建 BlockingCollection 且构造函数中指定了最大容量时，才会启用在达到最大容量时进行阻止的功能。
C#
{
for (int i = 0; i < itemsToAdd; i++)
{
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();
}
}
">using System;
using System.Collections.Concurrent;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
class Program
{
static void Main()
{
// Increase or decrease this value as desired.
int itemsToAdd = 500;
if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
{
int width = Math.Max(Console.BufferWidth, 80);
int height = Math.Max(Console.BufferHeight, itemsToAdd * 2 + 3);
// Preserve all the display output for Adds and Takes
Console.SetBufferSize(width, height);
}
// A bounded collection. Increase, decrease, or remove the
// maximum capacity argument to see how it impacts behavior.
var numbers = new BlockingCollection<int>(50);
// A simple blocking consumer with no cancellation.
Task.Run(() =>
{
int i = -1;
while (!numbers.IsCompleted)
{
try
{
i = numbers.Take();
}
catch (InvalidOperationException)
{
Console.WriteLine("Adding was completed!");
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.Run(() =>
{
for (int i = 0; i < itemsToAdd; i++)
{
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();
}
}
示例
第二个示例演示如何添加和取出项以便不会阻止操作。 如果没有任何项、已达到绑定集合的最大容量或已超过超时期限，TryAdd 或 TryTake 操作返回 false。 这样一来，线程可以暂时执行其他一些有用的工作，并在稍后再次尝试检索新项，或尝试添加先前无法添加的相同项。 该程序还演示如何在访问 BlockingCollection
bc, CancellationToken ct)
{
// IsCompleted == (IsAddingCompleted && Count == 0)
while (!bc.IsCompleted)
{
int nextItem = 0;
try
{
if (!bc.TryTake(out nextItem, 0, ct))
{
Console.WriteLine(" Take Blocked");
}
else
{
Console.WriteLine(" Take:{0}", nextItem);
}
}
catch (OperationCanceledException)
{
Console.WriteLine("Taking canceled.");
break;
}
// Slow down consumer just a little to cause
// collection to fill up faster, and lead to "AddBlocked"
Thread.SpinWait(500000);
}
Console.WriteLine("\r\nNo more items to take.");
}
static void NonBlockingProducer(BlockingCollection bc, CancellationToken ct)
{
int itemToAdd = 0;
bool success = false;
do
{
// Cancellation causes OCE. We know how to handle it.
try
{
// A shorter timeout causes more failures.
success = bc.TryAdd(itemToAdd, 2, ct);
}
catch (OperationCanceledException)
{
Console.WriteLine("Add loop canceled.");
// Let other threads know we're done in case
// they aren't monitoring the cancellation token.
bc.CompleteAdding();
break;
}
if (success)
{
Console.WriteLine(" Add:{0}", itemToAdd);
itemToAdd++;
}
else
{
Console.Write(" AddBlocked:{0} Count = {1}", itemToAdd.ToString(), bc.Count);
// Don't increment nextItem. Try again on next iteration.
//Do something else useful instead.
UpdateProgress(itemToAdd);
}
} while (itemToAdd < inputs);
// No lock required here because only one producer.
bc.CompleteAdding();
}
static void UpdateProgress(int i)
{
double percent = ((double)i / inputs) * 100;
Console.WriteLine("Percent complete: {0}", percent);
}
}
">using System;
using System.Collections.Concurrent;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
class ProgramWithCancellation
{
static int inputs = 2000;
static void Main()
{
// The token source for issuing the cancelation request.
var cts = new CancellationTokenSource();
// A blocking collection that can hold no more than 100 items at a time.
var numberCollection = new BlockingCollection<int>(100);
if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
{
int width = Math.Max(Console.BufferWidth, 80);
int height = Math.Max(Console.BufferHeight, 8000);
// Preserve all the display output for Adds and Takes
Console.SetBufferSize(width, height);
}
// The simplest UI thread ever invented.
Task.Run(() =>
{
if (Console.ReadKey(true).KeyChar == 'c')
{
cts.Cancel();
}
});
// Start one producer and one consumer.
Task t1 = Task.Run(() => NonBlockingConsumer(numberCollection, cts.Token));
Task t2 = Task.Run(() => NonBlockingProducer(numberCollection, cts.Token));
// Wait for the tasks to complete execution
Task.WaitAll(t1, t2);
cts.Dispose();
Console.WriteLine("Press the Enter key to exit.");
Console.ReadLine();
}
static void NonBlockingConsumer(BlockingCollection<int> bc, CancellationToken ct)
{
// IsCompleted == (IsAddingCompleted && Count == 0)
while (!bc.IsCompleted)
{
int nextItem = 0;
try
{
if (!bc.TryTake(out nextItem, 0, ct))
{
Console.WriteLine(" Take Blocked");
}
else
{
Console.WriteLine(" Take:{0}", nextItem);
}
}
catch (OperationCanceledException)
{
Console.WriteLine("Taking canceled.");
break;
}
// Slow down consumer just a little to cause
// collection to fill up faster, and lead to "AddBlocked"
Thread.SpinWait(500000);
}
Console.WriteLine("\r\nNo more items to take.");
}
static void NonBlockingProducer(BlockingCollection<int> bc, CancellationToken ct)
{
int itemToAdd = 0;
bool success = false;
do
{
// Cancellation causes OCE. We know how to handle it.
try
{
// A shorter timeout causes more failures.
success = bc.TryAdd(itemToAdd, 2, ct);
}
catch (OperationCanceledException)
{
Console.WriteLine("Add loop canceled.");
// Let other threads know we're done in case
// they aren't monitoring the cancellation token.
bc.CompleteAdding();
break;
}
if (success)
{
Console.WriteLine(" Add:{0}", itemToAdd);
itemToAdd++;
}
else
{
Console.Write(" AddBlocked:{0} Count = {1}", itemToAdd.ToString(), bc.Count);
// Don't increment nextItem. Try again on next iteration.
//Do something else useful instead.
UpdateProgress(itemToAdd);
}
} while (itemToAdd < inputs);
// No lock required here because only one producer.
bc.CompleteAdding();
}
static void UpdateProgress(int i)
{
double percent = ((double)i / inputs) * 100;
Console.WriteLine("Percent complete: {0}", percent);
}
}
另请参阅
- System.Collections.Concurrent
- BlockingCollection 概述
使用 foreach 删除 BlockingCollection 中的项
除了使用 Take 和 TryTake 方法从 BlockingCollectionforeach (For Each) 循环不同，此枚举器通过删除项来修改源集合，因此将其称作 转变枚举 或 耗用枚举。
示例
以下示例演示如何使用 foreach (For Each) 循环删除 BlockingCollection
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();
}
}
">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();
}
}
此示例将 foreach 循环与耗用线程中的 BlockingCollection
BlockingCollectionIEnumerable，因此无法保证顺序。 但是，在内部将 System.Collections.Concurrent.ConcurrentQueue
若要枚举集合而不对其进行修改，只需使用 foreach (For Each) 即可，无需使用 GetConsumingEnumerable 方法。 但是，务必要了解此类枚举表示的是某个精确时间点的集合快照。 如果其他线程在你执行循环的同时添加或删除项，则循环可能不会表示集合的实际状态。
请参阅
- System.Collections.Concurrent
- 并行编程
如何在 ConcurrentDictionary 中添加和删除项
本示例演示如何在 System.Collections.Concurrent.ConcurrentDictionary
ConcurrentDictionary
| 方法 | 何时使用… |
|---|---|
| AddOrUpdate | 需要为指定键添加新值，如果此键已存在，则需要替换其值。 |
| GetOrAdd | 需要检索指定键的现有值，如果此键不存在，则需要指定一个键/值对。 |
| TryAdd, TryGetValue, TryUpdate, TryRemove | 需要添加、获取、更新或移除键/值对，如果此键已存在或因任何其他原因导致尝试失败，则需执行某种备选操作。 |
示例
下面的示例使用两个 Task 实例将一些元素同时添加到 ConcurrentDictionary
TryAddCities(cityData)),
Task.Run(() => TryAddCities(cityData)));
static void TryAddCities(CityInfo[] cities)
{
for (var i = 0; i < cities.Length; ++i)
{
var (city, threadId) = (cities[i], Thread.CurrentThread.ManagedThreadId);
if (Cities.TryAdd(city.Name, city))
{
Console.WriteLine($"Thread={threadId}, added {city.Name}.");
}
else
{
Console.WriteLine($"Thread={threadId}, could not add {city.Name}, it was already added.");
}
}
}
// Enumerate collection from the app main thread.
// Note that ConcurrentDictionary is the one concurrent collection
// that does not support thread-safe enumeration.
foreach (var city in Cities)
{
Console.WriteLine($"{city.Key} has been added.");
}
AddOrUpdateWithoutRetrieving();
TryRemoveCity();
RetrieveValueOrAdd();
RetrieveAndUpdateOrAdd();
Console.WriteLine("Press any key.");
Console.ReadKey();
}
// This method shows how to add key-value pairs to the dictionary
// in scenarios where the key might already exist.
static void AddOrUpdateWithoutRetrieving()
{
// Sometime later. We receive new data from some source.
var ci = new CityInfo
{
Name = "Toronto",
Latitude = 43.716589M,
Longitude = -79.340686M,
RecentHighTemperatures = new int[] { 54, 59, 67, 82, 87, 55, -14 }
};
// Try to add data. If it doesn't exist, the object ci is added. If it does
// already exist, update existingVal according to the custom logic.
_ = Cities.AddOrUpdate(
ci.Name,
ci,
(cityName, existingCity) =>
{
// If this delegate is invoked, then the key already exists.
// Here we make sure the city really is the same city we already have.
if (ci != existingCity)
{
// throw new ArgumentException($"Duplicate city names are not allowed: {ci.Name}.");
}
// The only updatable fields are the temperature array and LastQueryDate.
existingCity.LastQueryDate = DateTime.Now;
existingCity.RecentHighTemperatures = ci.RecentHighTemperatures;
return existingCity;
});
// Verify that the dictionary contains the new or updated data.
Console.Write($"Most recent high temperatures for {ci.Name} are: ");
var temps = Cities[ci.Name].RecentHighTemperatures;
Console.WriteLine(string.Join(", ", temps));
}
// This method shows how to use data and ensure that it has been
// added to the dictionary.
static void RetrieveValueOrAdd()
{
var searchKey = "Caracas";
CityInfo retrievedValue = null;
try
{
retrievedValue = Cities.GetOrAdd(searchKey, GetDataForCity(searchKey));
}
catch (ArgumentException e)
{
Console.WriteLine(e.Message);
}
// Use the data.
if (retrievedValue != null)
{
Console.Write($"Most recent high temperatures for {retrievedValue.Name} are: ");
var temps = Cities[retrievedValue.Name].RecentHighTemperatures;
Console.WriteLine(string.Join(", ", temps));
}
}
// This method shows how to remove a value from the dictionary.
// If the value is unable to be removed, you can handle that by using the return
// boolean value from the .TryRemove function.
static void TryRemoveCity()
{
Console.WriteLine($"Total cities = {Cities.Count}");
var searchKey = "Milwaukee";
if (Cities.TryRemove(searchKey, out CityInfo retrievedValue))
{
Console.Write($"Most recent high temperatures for {retrievedValue.Name} are: ");
var temps = retrievedValue.RecentHighTemperatures;
Console.WriteLine(string.Join(", ", temps));
}
else
{
Console.WriteLine($"Unable to remove {searchKey}");
}
Console.WriteLine($"Total cities = {Cities.Count}");
}
// This method shows how to retrieve a value from the dictionary,
// when you expect that the key/value pair already exists,
// and then possibly update the dictionary with a new value for the key.
static void RetrieveAndUpdateOrAdd()
{
var searchKey = "Buenos Aires";
if (Cities.TryGetValue(searchKey, out CityInfo retrievedValue))
{
// Use the data.
Console.Write($"Most recent high temperatures for {retrievedValue.Name} are: ");
var temps = retrievedValue.RecentHighTemperatures;
Console.WriteLine(string.Join(", ", temps));
// Make a copy of the data. Our object will update its LastQueryDate automatically.
var newValue = new CityInfo
{
Name = retrievedValue.Name,
Latitude = retrievedValue.Latitude,
Longitude = retrievedValue.Longitude,
RecentHighTemperatures = retrievedValue.RecentHighTemperatures
};
// Replace the old value with the new value.
if (!Cities.TryUpdate(searchKey, newValue, retrievedValue))
{
// The data was not updated. Log error, throw exception, etc.
Console.WriteLine($"Could not update {retrievedValue.Name}");
}
}
else
{
// Add the new key and value. Here we call a method to retrieve
// the data. Another option is to add a default value here and
// update with real data later on some other thread.
var newValue = GetDataForCity(searchKey);
if (Cities.TryAdd(searchKey, newValue))
{
// Use the data.
Console.Write($"Most recent high temperatures for {newValue.Name} are: ");
var temps = newValue.RecentHighTemperatures;
Console.WriteLine(string.Join(", ", temps));
}
else
{
Console.WriteLine($"Unable to add data for {searchKey}");
}
}
}
// Assume this method knows how to find long/lat/temp info for any specified city.
static CityInfo GetDataForCity(string name) => name switch
{
"Caracas" => new CityInfo
{
Name = "Caracas",
Longitude = 10.5M,
Latitude = -66.916667M,
RecentHighTemperatures = new int[] { 91, 89, 91, 91, 87, 90, 91 }
},
"Buenos Aires" => new CityInfo
{
Name = "Buenos Aires",
Longitude = -34.61M,
Latitude = -58.369997M,
RecentHighTemperatures = new int[] { 80, 86, 89, 91, 84, 86, 88 }
},
_ => throw new ArgumentException($"Cannot find any data for {name}")
};
}
}
">using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
namespace DictionaryHowTo
{
// The type of the Value to store in the dictionary.
class CityInfo : IEqualityComparer<CityInfo>
{
public string Name { get; set; }
public DateTime LastQueryDate { get; set; } = DateTime.Now;
public decimal Longitude { get; set; } = decimal.MaxValue;
public decimal Latitude { get; set; } = decimal.MaxValue;
public int[] RecentHighTemperatures { get; set; } = new int[] { 0 };
public bool Equals(CityInfo x, CityInfo y)
=> (x.Name, x.Longitude, x.Latitude) ==
(y.Name, y.Longitude, y.Latitude);
public int GetHashCode(CityInfo cityInfo) =>
cityInfo?.Name.GetHashCode() ?? throw new ArgumentNullException(nameof(cityInfo));
}
class Program
{
static readonly ConcurrentDictionary<string, CityInfo> Cities =
new ConcurrentDictionary<string, CityInfo>(StringComparer.OrdinalIgnoreCase);
static async Task Main()
{
CityInfo[] cityData =
{
new CityInfo { Name = "Boston", Latitude = 42.358769m, Longitude = -71.057806m, RecentHighTemperatures = new int[] { 56, 51, 52, 58, 65, 56,53} },
new CityInfo { Name = "Miami", Latitude = 25.780833m, Longitude = -80.195556m, RecentHighTemperatures = new int[] { 86,87,88,87,85,85,86 } },
new CityInfo { Name = "Los Angeles", Latitude = 34.05m, Longitude = -118.25m, RecentHighTemperatures = new int[] { 67,68,69,73,79,78,78 } },
new CityInfo { Name = "Seattle", Latitude = 47.609722m, Longitude = -122.333056m, RecentHighTemperatures = new int[] { 49,50,53,47,52,52,51 } },
new CityInfo { Name = "Toronto", Latitude = 43.716589m, Longitude = -79.340686m, RecentHighTemperatures = new int[] { 53,57, 51,52,56,55,50 } },
new CityInfo { Name = "Mexico City", Latitude = 19.432736m, Longitude = -99.133253m, RecentHighTemperatures = new int[] { 72,68,73,77,76,74,73 } },
new CityInfo { Name = "Rio de Janeiro", Latitude = -22.908333m, Longitude = -43.196389m, RecentHighTemperatures = new int[] { 72,68,73,82,84,78,84 } },
new CityInfo { Name = "Quito", Latitude = -0.25m, Longitude = -78.583333m, RecentHighTemperatures = new int[] { 71,69,70,66,65,64,61 } },
new CityInfo { Name = "Milwaukee", Latitude = -43.04181m, Longitude = -87.90684m, RecentHighTemperatures = new int[] { 32,47,52,64,49,44,56 } }
};
// Add some key/value pairs from multiple threads.
await Task.WhenAll(
Task.Run(() => TryAddCities(cityData)),
Task.Run(() => TryAddCities(cityData)));
static void TryAddCities(CityInfo[] cities)
{
for (var i = 0; i < cities.Length; ++i)
{
var (city, threadId) = (cities[i], Thread.CurrentThread.ManagedThreadId);
if (Cities.TryAdd(city.Name, city))
{
Console.WriteLine($"Thread={threadId}, added {city.Name}.");
}
else
{
Console.WriteLine($"Thread={threadId}, could not add {city.Name}, it was already added.");
}
}
}
// Enumerate collection from the app main thread.
// Note that ConcurrentDictionary is the one concurrent collection
// that does not support thread-safe enumeration.
foreach (var city in Cities)
{
Console.WriteLine($"{city.Key} has been added.");
}
AddOrUpdateWithoutRetrieving();
TryRemoveCity();
RetrieveValueOrAdd();
RetrieveAndUpdateOrAdd();
Console.WriteLine("Press any key.");
Console.ReadKey();
}
// This method shows how to add key-value pairs to the dictionary
// in scenarios where the key might already exist.
static void AddOrUpdateWithoutRetrieving()
{
// Sometime later. We receive new data from some source.
var ci = new CityInfo
{
Name = "Toronto",
Latitude = 43.716589M,
Longitude = -79.340686M,
RecentHighTemperatures = new int[] { 54, 59, 67, 82, 87, 55, -14 }
};
// Try to add data. If it doesn't exist, the object ci is added. If it does
// already exist, update existingVal according to the custom logic.
_ = Cities.AddOrUpdate(
ci.Name,
ci,
(cityName, existingCity) =>
{
// If this delegate is invoked, then the key already exists.
// Here we make sure the city really is the same city we already have.
if (ci != existingCity)
{
// throw new ArgumentException($"Duplicate city names are not allowed: {ci.Name}.");
}
// The only updatable fields are the temperature array and LastQueryDate.
existingCity.LastQueryDate = DateTime.Now;
existingCity.RecentHighTemperatures = ci.RecentHighTemperatures;
return existingCity;
});
// Verify that the dictionary contains the new or updated data.
Console.Write($"Most recent high temperatures for {ci.Name} are: ");
var temps = Cities[ci.Name].RecentHighTemperatures;
Console.WriteLine(string.Join(", ", temps));
}
// This method shows how to use data and ensure that it has been
// added to the dictionary.
static