碼上快樂

相關內容    簡體    繁體

線程安全集合

本文轉載自   查看原文   2020-02-01 11:43   751    多線程/ 並發編程

>>返回《C# 並發編程》

1. 簡介

  • 不可變集合
    • 不可變集合之間通常共享了大部分存儲空間，因此其實浪費並不大
    • 因為是無法修改的，所以是線程安全的
  • 線程安全集合
    • 可同時被多個線程修改的可變集合
      • 線程安全集合混合使用了細粒度鎖定和無鎖技術，以確保線程被阻塞的時間最短
        • 通常情況下是根本不阻塞
    • 對很多線程安全集合進行枚舉操作時，內部創建了該集合的一個快照(snapshot)，並對這個快照進行枚舉操作。
    • 線程安全集合的主要優點是多個線程可以安全地對其進行訪問，而代碼只會被阻塞很短的時間，或根本不阻塞。

下面對常用的屬於不可變集合和線程安全集合類型的，特定數據結構的集合進行說明。

2. 不可變棧和隊列

不可變集合采用的模式是返回一個修改過的集合，原始的集合引用是不變化的。

  • 這意味着，如果引用了特定的不可變集合的實例，它是不會變化的。
var stack = ImmutableStack<int>.Empty;
stack = stack.Push(13);
var biggerStack = stack.Push(7);
// 先顯示“7”，接着顯示“13”。 
foreach (var item in biggerStack)
    Console.WriteLine($"biggerStack {item}");

// 只顯示“13”。
foreach (var item in stack)
    Console.WriteLine($"stack {item}");

輸出:

biggerStack 7
biggerStack 13
stack 13

兩個棧實際上在內部共享了存儲項目 13 的內存。

  • 這種實現方式的效率很高，並且可以很方便地創建當前狀態的快照
  • 每個不可變集合的實例都是絕對線程安全的

ImmutableQueue 使用方法類似。

  • 不可變集合的一個實例是永遠不改變的。
  • 因為不會改變，所以是絕對線程安全的。
  • 對不可變集合使用修改方法時，返回修改后的集合。
  • 不可變集合非常適用於共享狀態，但不適合用來做交換數據的通道。

3. 不可變列表

不可變列表的內部是用二叉樹組織數據的。這么做是為了讓不可變列表的實例之間共享的內存最大化。

  • 這導致 ImmutableList<T> 和 List<T> 在常用操作上有性能上的差別(參見下表)。
操 作 List<T> ImmutableList<T>
Add 平攤 O(1) O(log N)
Insert O(N) O(log N)
RemoveAt O(N) O(log N)
Item[index] O(1) O(log N)

不可變列表確實可以使用index獲取數據項，但需要注意性能問題。不能簡單地用它來替代 List<T>。

  • 這意味着應該盡量使用 foreach 而不是用 for

4. 不可變Set集合

  • ImmutableHashSet<T>
    • 是一個不含重復元素的集合
  • ImmutableSortedSet<T>
    • 是一個已排序的不含重復元素的集合
  • 都有相似的接口
//ImmutableHashSet
var hashSet = ImmutableHashSet<int>.Empty;
hashSet = hashSet.Add(13);
hashSet = hashSet.Add(7);
// 顯示“7”和“13”，次序不確定。 
foreach (var item in hashSet)
    Console.Write(item + " ");

System.Console.WriteLine();
hashSet = hashSet.Remove(7);

//ImmutableSortedSet
var sortedSet = ImmutableSortedSet<int>.Empty;
sortedSet = sortedSet.Add(13);
sortedSet = sortedSet.Add(7);
// 先顯示“7”，接着顯示“13”。 
foreach (var item in sortedSet)
    Console.Write(item + " ");

var smallestItem = sortedSet[0];
// smallestItem == 7
sortedSet = sortedSet.Remove(7);

輸出:

7 13 
7 13
操 作 ImmutableHashSet<T> ImmutableSortedSet<T>
Add O(log N) O(log N)
Remove O(log N) O(log N)
Item[index] 不可用 O(log N)

ImmutableSortedSet 索引操作的時間復雜度是 O(log N)，而不是 O(1)，這跟 上節中 ImmutableList<T> 的情況類似。

  • 這意味着它們適用同樣的警告:使用 ImmutableSortedSet<T>時，應該盡量用 foreach 而不是用 for 。

可以先快速地以可變方式構建，然后轉換成不可變集合。

5. 不可變字典

  • ImmutableDictionary<TKey,TValue>
  • ImmutableSortedDictionar y<TKey,TValue>
//ImmutableDictionary
var dictionary = ImmutableDictionary<int, string>.Empty;
dictionary = dictionary.Add(10, "Ten");
dictionary = dictionary.Add(21, "Twenty-One");
dictionary = dictionary.SetItem(10, "Diez");
// 顯示“10Diez”和“21Twenty-One”，次序不確定。 
foreach (var item in dictionary)
    Console.WriteLine(item.Key + ":" + item.Value);

var ten = dictionary[10]; // ten == "Diez"
dictionary = dictionary.Remove(21);

//ImmutableSortedDictionary
var sortedDictionary = ImmutableSortedDictionary<int, string>.Empty; sortedDictionary = sortedDictionary.Add(10, "Ten");
sortedDictionary = sortedDictionary.Add(21, "Twenty-One");
sortedDictionary = sortedDictionary.SetItem(10, "Diez");
// 先顯示“10Diez”，接着顯示“21Twenty-One”。 
foreach (var item in sortedDictionary)
    Console.WriteLine(item.Key + ":" + item.Value);

ten = sortedDictionary[10];
// ten == "Diez"
sortedDictionary = sortedDictionary.Remove(21);

輸出:

10:Diez
21:Twenty-One
10:Diez
21:Twenty-One
操 作	I
操 作 ImmutableDictionary<TK,TV> ImmutableSortedDictionary<TK,TV>
Add O(log N) O(log N)
SetItem O(log N) O(log N)
Item[key] O(log N) O(log N)
Remove O(log N) O(log N)

6. 線程安全字典

var dictionary = new ConcurrentDictionary<int, string>(); 
var newValue = dictionary.AddOrUpdate(0,
key => "Zero",
(key, oldValue) => "Zero");

AddOrUpdate 方法有些復雜，這是因為這個方法必須執行多個步驟，具體步驟取決於並發字典的當前內容。

  • 方法的第一個參數是鍵
  • 第二個參數是一個委托，它把鍵(本例中為 0)轉換成添加到字典的值(本例中為“Zero”)
    • 只有當字典中沒有這個鍵時，這個委托才會運行。
  • 第三個參數也是一個委托，它把鍵(0)和原來的值轉換成字典中修改后的值
    (“Zero”)。
    • 只有當字典中已經存在這個鍵時，這個委托才會運行。
  • AddOrUpdate return 這個鍵對應的新值(與其中一個委托返回的值相同)。

AddOrUpdate 可能要多次調用其中一個(或兩個)委托。這種情況很少，但確實會發生。

  • 因此這些委托必須簡單、快速，並且不能有副作用
  • 這些委托只能創建新的值，不能修改程序中其他變量
  • 這個原則適用於所有 ConcurrentDictionary<TKey,TValue> 的方法所使用的委托
// 使用與前面一樣的“字典”。
string currentValue;
bool keyExists = dictionary.TryGetValue(0, out currentValue);

// 使用與前面一樣的“字典”。
string removedValue;
bool keyExisted = dictionary.TryRemove(0, out removedValue);

  • 如果多個線程讀寫一個共享集合， 使用 ConcurrentDictrionary<TKey,TValue> 是最合適的

  • 如果不會頻繁修改(很少修改)， 那更適合使用 ImmutableDictionary<TKey, TValue>。

  • 如果一些線程只添加元素，另一些線程只移除元素，那最好使用生產者/消費者集合。

7. 阻塞隊列

  • GetConsumingEnumerable 會阻塞線程
  • CommpleteAdding 方法執行后所有被 GetConsumingEnumerable 阻塞的線程開始執行
  • 每個元素只會被消費一次
private static readonly BlockingCollection<int> _blockingQueue = new BlockingCollection<int>();
public static async Task BlockingCollectionSP()
{
    Action consumerAction = () =>
     {
        Console.WriteLine($"started print({Thread.CurrentThread.ManagedThreadId}).");
        // 先顯示“7”，后顯示“13”。
        foreach (var item in _blockingQueue.GetConsumingEnumerable())
        {
             Console.WriteLine($"print({Thread.CurrentThread.ManagedThreadId}) {item}");
        }
        Console.WriteLine($"ended print({Thread.CurrentThread.ManagedThreadId}).");
     };
    Task task1 = Task.Run(consumerAction);
    Task task2 = Task.Run(consumerAction);
    Task task3 = Task.Run(consumerAction);

    _blockingQueue.Add(7);
    System.Console.WriteLine($"added 7.");
    _blockingQueue.Add(13);
    System.Console.WriteLine($"added 13.");
    _blockingQueue.CompleteAdding();
    System.Console.WriteLine("CompleteAdding.");

    try
    {
        _blockingQueue.Add(15);
    }
    catch (Exception ex)
    {
        System.Console.WriteLine($"{ex.GetType().Name}:{ex.Message}");
    }

    await Task.WhenAll(task1, task2, task3);
}

輸出:

started print(4).
started print(3).
started print(6).
added 7.
added 13.
CompleteAdding.
ended print(6).
InvalidOperationException:The collection has been marked as complete with regards to additions.
print(4) 7
ended print(4).
print(3) 13
ended print(3).

8. 阻塞棧和包

  • 在默認情況下，.NET 中的 BlockingCollection<T> 用作阻塞隊列，但它也可以作為任何類型的生產者/消費者集合。
  • BlockingCollection<T> 實際上是對線程安全集合進行了封裝， 實現了 IProducerConsumerCollection<T> 接口。
    • 因此可以在創建 BlockingCollection<T> 實例時指明規則
BlockingCollection<int> _blockingStack = new BlockingCollection<int>( new ConcurrentStack<int>());
BlockingCollection<int> _blockingBag = new BlockingCollection<int>( new ConcurrentBag<int>());

替換到阻塞隊列示例代碼中試試。

9. 異步隊列

public static async Task BufferBlockPS()
{
    BufferBlock<int> _asyncQueue = new BufferBlock<int>();
    Func<Task> concurrentConsumerAction = async () =>
     {
         while (true)
         {
             int item;
             try
             {
                 item = await _asyncQueue.ReceiveAsync();
             }
             catch (InvalidOperationException)
             {
                 System.Console.WriteLine($"exit({Thread.CurrentThread.ManagedThreadId}).");
                 break;
             }
             Console.WriteLine($"print({Thread.CurrentThread.ManagedThreadId}) {item}");
         }
     };
    Func<Task> consumerAction = async () =>
    {
        try
        {
            // 先顯示“7”，后顯示“13”。 單線程可用
            while (await _asyncQueue.OutputAvailableAsync())
            {
                Console.WriteLine($"print({Thread.CurrentThread.ManagedThreadId}) {await _asyncQueue.ReceiveAsync()}");
            }
        }
        catch (Exception ex)
        {
            System.Console.WriteLine($"{ex.GetType().Name}({Thread.CurrentThread.ManagedThreadId}):{ex.Message}");
        }

    };

    Task t1 = consumerAction();
    Task t2 = consumerAction();

    // Task t1 = concurrentConsumerAction();
    // Task t2 = concurrentConsumerAction();

    // 生產者代碼
    await _asyncQueue.SendAsync(7);
    await _asyncQueue.SendAsync(13);
    await _asyncQueue.SendAsync(15);
    System.Console.WriteLine("Added 7 13 15.");
    _asyncQueue.Complete();

    await Task.WhenAll(t1, t2);
}

輸出:

Added 7 13 15.
print(4) 7
print(6) 13
print(4) 15
InvalidOperationException(3):The source completed without providing data to receive.

10. 異步棧和包

Nito.AsyncEx 庫

AsyncCollection<int> _asyncStack = new AsyncCollection<int>( new ConcurrentStack<int>());
AsyncCollection<int> _asyncBag = new AsyncCollection<int>( new ConcurrentBag<int>());

11. 阻塞/異步隊列

在阻塞隊列中已經介紹了BufferBlock<T>

這里介紹 ActionBlock<int>

public static async Task ActionBlockPS()
{
    ActionBlock<int> queue = new ActionBlock<int>(u => Console.WriteLine($"print({Thread.CurrentThread.ManagedThreadId}) {u}"));

    // 異步的生產者代碼
    await queue.SendAsync(7);
    await queue.SendAsync(13);
    System.Console.WriteLine("Added async.");
    // 同步的生產者代碼 
    queue.Post(15);
    queue.Post(17);
    System.Console.WriteLine("Added sync.");
    queue.Complete();
    System.Console.WriteLine($"Completed({Thread.CurrentThread.ManagedThreadId}).");
}

輸出:

Added async.
Added sync.
Completed(1).
print(3) 7
print(3) 13
print(3) 15
print(3) 17


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 線程安全集合 .Net 線程安全集合 C# 線程安全集合 Concurrent下的線程安全集合 C# 4.0 之線程安全集合篇 C# 中的線程安全集合類 .NET(C#)：線程安全集合的阻塞BlockingCollection的使用 C# 多線程安全集合類 C# 線程安全集合類 最全java多線程總結3——了解阻塞隊列和線程安全集合不
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM