.NET multi-threaded programming—concurrent collections

• 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 (m.sbmmt.com)!