.NET multi-threaded programming—concurrent collections

Concurrent Collection

1 Why use concurrent collection?

The main reasons are as follows:

• The classic lists, collections and arrays provided in the System.Collections and System.Collections.Generic namespaces are not thread-safe Without a synchronization mechanism, they are not suitable for accepting concurrent instructions to add and remove elements.

• Using the above classic collections in concurrent code requires complex synchronization management, which is very inconvenient to use.

• Using complex synchronization mechanisms will greatly reduce performance.

• The new collections provided by NET Framework 4 minimize the number of times you need to use locks. These new collections avoid the use of mutually exclusive heavyweight locks by using compare-and-swap (CAS) instructions and memory barriers. This guarantees performance.

Note: Compared with classic collections, concurrent collections will have greater overhead, so using concurrent collections in serial code is meaningless and will only increase additional overhead and run speed. Slower than accessing the classic collection.

2 Concurrent collection

1) ConcurrentQueue: thread-safe first-in-first-out (FIFO) collection

Main method:

• Enqueue(T item);Adds the object to the end of the collection.

• TryDequeue(out T result); Try to remove and return the object at the beginning of the collection. The return value indicates whether the operation was successful.

• TryPeek(out T result); Try to return the object at the beginning of the collection without removing it. The return value indicates whether the operation was successful.

Note:

• ConcurrentQueue is completely lock-free, but when a CAS operation fails and faces resource contention, it may spin and Retry the operation.

• ConcurrentQueue is a FIFO collection. In some situations that have nothing to do with the order of entry and exit, try not to use ConcurrentQueue.

2) ConcurrentStack: Thread-safe last-in-first-out (LIFO) collection

Main methods and properties:

• Push (T item);Inserts the object at the top of the collection.

• TryPop(out T result); Try to pop and return the object at the top of the collection. The return value indicates whether the operation is successful.

• TryPeek(out T result); Try to return the object at the beginning of the collection without removing it. The return value indicates whether the operation was successful.

• IsEmpty { get; } Indicates whether the collection is empty.

• PushRange(T[] items);Insert multiple objects at the top of the collection.

• TryPopRange(T[] items); Pops multiple elements at the top, and the return result is the number of popped elements.

Note:

• Similar to ConcurrentQueue, ConcurrentStack is completely lock-free, but when the CAS operation fails and faces resource contention, it may Will spin and retry the operation.

• Get whether the collection contains elements using the IsEmpty property instead of judging whether the Count property is greater than zero. Calling Count is more expensive than calling IsEmpty.

• When using PushRange(T[] items) and TryPopRange(T[] items), pay attention to the extra overhead and extra memory consumption caused by buffering.

3) ConcurrentBag: an unordered collection with repeatable elements

Main methods and attributes:

• TryPeek(out T result); attempts to return an object from the collection without removing the object. The return value indicates whether the object was successfully obtained.

• TryTake(out T result); Try to return an object from the collection and remove the object. The return value indicates whether the object was successfully obtained.

• Add(T item); Add the object to the collection.

• IsEmpty { get; }The explanation is the same as ConcurrentStack

##Instructions:

• ConcurrentBag is for each The thread accessing the collection maintains a local queue and, when possible, accesses the local queue in a lock-free manner.

• ConcurrentBag is very efficient when adding and removing elements in the same thread.

• Because ConcurrentBag sometimes requires locks, it is very inefficient in scenarios where producer threads and consumer threads are completely separated.

• ConcurrentBag's call to IsEmpty is very expensive because it requires temporarily acquiring all the locks of this unordered group.

4) BlockingCollection: A thread-safe collection that implements

System.Collections.Concurrent.IProducerConsumerCollection, providing blocking and limiting functions

Mainly Methods and properties:

• BlockingCollection(int boundedCapacity); boundedCapacity represents the collection limit size.

• CompleteAdding(); Marks the BlockingCollection instance as no longer accepting any additions.

• IsCompleted { get; } Whether this collection has been marked as completed and empty.

• GetConsumingEnumerable();Removes from the collection and returns the removed element

• Add(T item);Adds an element to the collection.

• TryTake(T item, int millisecondsTimeout, CancellationToken cancellationToken);

Instructions:

• Use BlockingCollection( ) constructor instantiates BlockingCollection, which means boundedCapacity is not set, then boundedCapacity is the default value: int.MaxValue.

• Bound: Use BlockingCollection(int boundedCapacity) to set the value of boundedCapacity. When the collection capacity reaches this value, the thread that adds elements to BlockingCollection will be blocked until an element is deleted. .

The bounding function controls the maximum size of the collection in memory, which is very useful when a large number of elements need to be processed.

• By default, BlockingCollection encapsulates a ConcurrentQueue. You can specify a concurrent collection that implements the IProducerConsumerCollection interface in the constructor, including: ConcurrentStack and ConcurrentBag.

• Using this collection involves the risk of waiting indefinitely, so it is more convenient to use TryTake because TryTake provides timeout control and an item can be removed from the collection within a specified time. is true; otherwise, it is false.

5) ConcurrentDictionary: A thread-safe collection of key-value pairs that can be accessed by multiple threads simultaneously.

Main method

AddOrUpdate(TKey key, TValue addValue, Func updateValueFactory); If the specified key does not exist yet, add the key/value pair to the dictionary Medium; if the specified key already exists, updates the key/value pair in the dictionary.

• GetOrAdd(TKey key, TValue value); If the specified key does not exist yet, adds the key/value pair to the dictionary.

• TryRemove(TKey key, out TValue value);Try to remove from the dictionary and return the value with the specified key.

• TryUpdate(TKey key, TValue newValue, TValue comparisonValue); Compares the existing value of the specified key with the specified value. If equal, updates the key with the third value.

Note:

• ConcurrentDictionary is completely lock-free for read operations. ConcurrentDictionary uses fine-grained locks when multiple tasks or threads add elements or modify data to it. Using fine-grained locks only locks the part that really needs to be locked, not the entire dictionary.

6) IProducerConsumerCollection: Defines methods for producers/consumers to operate thread-safe collections. This interface provides a unified representation (for producer/consumer collections) so that higher level abstractions such as System.Collections.Concurrent.BlockingCollection can use collections as the underlying storage mechanism.

3. Commonly used patterns

1) Parallel producer-consumer pattern

Definition:

The producer and consumer are in this pattern Two types of object models, the consumer depends on the results of the producer, while the producer generates the results, the consumer uses the results.

Figure 1 Parallel producer-consumer model

Description:

• Concurrent collections are used here pattern is a good fit because concurrent collections support parallel operations on objects in this pattern.

• If you do not use concurrent collections, you must add a synchronization mechanism, which will make the program more complex, difficult to maintain and understand, and greatly reduce performance.

• The above picture is a schematic diagram of the producer-consumer model. The vertical axis is the timeline. The producer and the consumer are not on the same timeline, but they overlap, which is intended to show that The generator first produces results, and then the consumer actually uses the data generated by the generator.

2) Pipeline pattern

Definition:

The pipeline consists of multiple stages, each stage consists of a series of producers and consumers . Generally speaking, the previous stage is the generator of the later stage; relying on the buffer queue between two adjacent stages, each stage can be executed concurrently.

Figure 2 Parallel pipeline mode

Description:

• BlockingCollection is often used as a buffer tank Zone queue.

• The speed of the pipeline is approximately equal to the speed of the slowest stage of the pipeline.

• The above picture is a schematic diagram of the pipeline mode. The previous stage is the generator of the later stage. The simplest and basic pipeline mode is shown here. More complex modes can be considered as each Each stage includes more processing of the data.

4 How to use

Only take ConcurrentBag and BlockingCollection as examples, other concurrent collections are similar.

ConcurrentBag

List<string> list = ......
ConcurrentBag<string> bags = new ConcurrentBag<string>();
Parallel.ForEach(list, (item) => 
{
    //对list中的每个元素进行处理然后，加入bags中
    bags.Add(itemAfter);
});

BlockingCollection—Producer Consumer Pattern

public static void Execute()
{
            //调用Invoke，使得生产者任务和消费者任务并行执行
            //Producer方法和Customer方法在Invoke中的参数顺序任意，不论何种顺序都会获得正确的结果
            Parallel.Invoke(()=>Customer(),()=>Producer());
            Console.WriteLine(string.Join(",",customerColl));
}
//生产者集合
private static BlockingCollection<int> producerColl = new BlockingCollection<int>();
 //消费者集合
private static BlockingCollection<string> customerColl = new BlockingCollection<string>();
public static void Producer()
{
            //循环将数据加入生成者集合
            for (int i = 0; i < 100; i++)
            {
                producerColl.Add(i);
            }
            //设置信号，表明不在向生产者集合中加入新数据
            //可以设置更加复杂的通知形式，比如数据量达到一定值且其中的数据满足某一条件时就设置完成添加
            producerColl.CompleteAdding();
}
public static void Customer()
{
            //调用IsCompleted方法，判断生产者集合是否在添加数据，是否还有未"消费"的数据
            //注意不要使用IsAddingCompleted，IsAddingCompleted只表明集合标记为已完成添加，而不能说明其为空
            //而IsCompleted为ture时，那么IsAddingCompleted为ture且集合为空
            while (!producerColl.IsCompleted)
            {
                //调用Take或TryTake "消费"数据,消费一个，移除一个
                //TryAdd的好处是提供超时机制
                customerColl.Add(string.Format("消费：{0}", producerColl.Take()));
            }
}

The above is the content of .NET multi-thread programming - concurrent collection. For more related content, please pay attention to the PHP Chinese website (www.php.cn)!