程序師世界是廣大編程愛好者互助、分享、學習的平台,程序師世界有你更精彩!
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫知識
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
 程式師世界 >> 編程語言 >> .NET網頁編程 >> C# >> C#入門知識 >> C#並行編程-並發集合,

C#並行編程-並發集合,

編輯:C#入門知識

C#並行編程-並發集合,


菜鳥學習並行編程,參考《C#並行編程高級教程.PDF》,如有錯誤,歡迎指正。

背景

基於任務的程序設計、命令式數據並行和任務並行都要求能夠支持並發更新的數組、列表和集合。

在.NET Framework 4 以前,為了讓共享的數組、列表和集合能夠被多個線程更新,需要添加復雜的代碼來同步這些更新操作。

如您需要編寫一個並行循環,這個循環以無序的方式向一個共享集合中添加元素,那麼必須加入一個同步機制來保證這是一個線程安全的集合。

System.Collenctions和System.Collenctions.Generic 名稱空間中所提供的經典列表、集合和數組的線程都不是安全的,不能接受並發請求,因此需要對相應的操作方法執行串行化。

下面看代碼,代碼中並沒有實現線程安全和串行化:

class Program { private static object o = new object(); private static List<Product> _Products { get; set; } /* coder:釋迦苦僧 * 代碼中 創建三個並發線程 來操作_Products 集合 * System.Collections.Generic.List 這個列表在多個線程訪問下,不能保證是安全的線程,所以不能接受並發的請求,我們必須對ADD方法的執行進行串行化 */ static void Main(string[] args) { _Products = new List<Product>(); /*創建任務 t1 t1 執行 數據集合添加操作*/ Task t1 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t2 t2 執行 數據集合添加操作*/ Task t2 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t3 t3 執行 數據集合添加操作*/ Task t3 = Task.Factory.StartNew(() => { AddProducts(); }); Task.WaitAll(t1, t2, t3); Console.WriteLine(_Products.Count); Console.ReadLine(); } /*執行集合數據添加操作*/ static void AddProducts() { Parallel.For(0, 1000, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; _Products.Add(product); }); } } class Product { public string Name { get; set; } public string Category { get; set; } public int SellPrice { get; set; } } View Code

代碼中開啟了三個並發操作,每個操作都向集合中添加1000條數據,在沒有保障線程安全和串行化的運行下,實際得到的數據並沒有3000條,結果如下:
 

為此我們需要采用Lock關鍵字,來確保每次只有一個線程來訪問  _Products.Add(product); 這個方法,代碼如下:

class Program { private static object o = new object(); private static List<Product> _Products { get; set; } /* coder:釋迦苦僧 * 代碼中 創建三個並發線程 來操作_Products 集合 * System.Collections.Generic.List 這個列表在多個線程訪問下,不能保證是安全的線程,所以不能接受並發的請求,我們必須對ADD方法的執行進行串行化 */ static void Main(string[] args) { _Products = new List<Product>(); /*創建任務 t1 t1 執行 數據集合添加操作*/ Task t1 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t2 t2 執行 數據集合添加操作*/ Task t2 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t3 t3 執行 數據集合添加操作*/ Task t3 = Task.Factory.StartNew(() => { AddProducts(); }); Task.WaitAll(t1, t2, t3); Console.WriteLine("當前數據量為:" + _Products.Count); Console.ReadLine(); } /*執行集合數據添加操作*/ static void AddProducts() { Parallel.For(0, 1000, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; lock (o) { _Products.Add(product); } }); } } class Product { public string Name { get; set; } public string Category { get; set; } public int SellPrice { get; set; } } View Code

但是鎖的引入,帶來了一定的開銷和性能的損耗,並降低了程序的擴展性,在並發編程中顯然不適用。

 

System.Collections.Concurrent

.NET Framework 4提供了新的線程安全和擴展的並發集合,它們能夠解決潛在的死鎖問題和競爭條件問題,因此在很多復雜的情形下它們能夠使得並行代碼更容易編寫,這些集合盡可能減少需要使用鎖的次數,從而使得在大部分情形下能夠優化為最佳性能,不會產生不必要的同步開銷。

需要注意的是:

線程安全並不是沒有代價的,比起System.Collenctions和System.Collenctions.Generic命名空間中的列表、集合和數組來說,並發集合會有更大的開銷。因此,應該只在需要從多個任務中並發訪問集合的時候才使用並發幾個,在串行代碼中使用並發集合是沒有意義的,因為它們會增加無謂的開銷。

 

為此,在.NET Framework中提供了System.Collections.Concurrent新的命名空間可以訪問用於解決線程安全問題,通過這個命名空間能訪問以下為並發做好了准備的集合。

1.BlockingCollection 與經典的阻塞隊列數據結構類似,能夠適用於多個任務添加和刪除數據,提供阻塞和限界能力。

2.ConcurrentBag 提供對象的線程安全的無序集合

3.ConcurrentDictionary  提供可有多個線程同時訪問的鍵值對的線程安全集合

4.ConcurrentQueue   提供線程安全的先進先出集合

5.ConcurrentStack   提供線程安全的後進先出集合

這些集合通過使用比較並交換和內存屏障等技術,避免使用典型的互斥重量級的鎖,從而保證線程安全和性能。

 

ConcurrentQueue  

ConcurrentQueue 是完全無鎖的,能夠支持並發的添加元素,先進先出。下面貼代碼,詳解見注釋:

class Program { private static object o = new object(); /*定義 Queue*/ private static Queue<Product> _Products { get; set; } private static ConcurrentQueue<Product> _ConcurrenProducts { get; set; } /* coder:釋迦苦僧 * 代碼中 創建三個並發線程 來操作_Products 和 _ConcurrenProducts 集合,每次添加 10000 條數據 查看 一般隊列Queue 和 多線程安全下的隊列ConcurrentQueue 執行情況 */ static void Main(string[] args) { Thread.Sleep(1000); _Products = new Queue<Product>(); Stopwatch swTask = new Stopwatch(); swTask.Start(); /*創建任務 t1 t1 執行 數據集合添加操作*/ Task t1 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t2 t2 執行 數據集合添加操作*/ Task t2 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t3 t3 執行 數據集合添加操作*/ Task t3 = Task.Factory.StartNew(() => { AddProducts(); }); Task.WaitAll(t1, t2, t3); swTask.Stop(); Console.WriteLine("List<Product> 當前數據量為:" + _Products.Count); Console.WriteLine("List<Product> 執行時間為:" + swTask.ElapsedMilliseconds); Thread.Sleep(1000); _ConcurrenProducts = new ConcurrentQueue<Product>(); Stopwatch swTask1 = new Stopwatch(); swTask1.Start(); /*創建任務 tk1 tk1 執行 數據集合添加操作*/ Task tk1 = Task.Factory.StartNew(() => { AddConcurrenProducts(); }); /*創建任務 tk2 tk2 執行 數據集合添加操作*/ Task tk2 = Task.Factory.StartNew(() => { AddConcurrenProducts(); }); /*創建任務 tk3 tk3 執行 數據集合添加操作*/ Task tk3 = Task.Factory.StartNew(() => { AddConcurrenProducts(); }); Task.WaitAll(tk1, tk2, tk3); swTask1.Stop(); Console.WriteLine("ConcurrentQueue<Product> 當前數據量為:" + _ConcurrenProducts.Count); Console.WriteLine("ConcurrentQueue<Product> 執行時間為:" + swTask1.ElapsedMilliseconds); Console.ReadLine(); } /*執行集合數據添加操作*/ static void AddProducts() { Parallel.For(0, 30000, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; lock (o) { _Products.Enqueue(product); } }); } /*執行集合數據添加操作*/ static void AddConcurrenProducts() { Parallel.For(0, 30000, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; _ConcurrenProducts.Enqueue(product); }); } } class Product { public string Name { get; set; } public string Category { get; set; } public int SellPrice { get; set; } } View Code

需要注意的是,代碼中的輸出時間並不能夠完全正確的展示出並發代碼下的ConcurrentQueue性能,采用ConcurrentQueue在一定程度上也帶來了損耗,如下圖所示:

class Program { private static object o = new object(); private static ConcurrentQueue<Product> _ConcurrenProducts { get; set; } /* coder:釋迦苦僧 * ConcurrentQueue 下的 TryPeek 和 TryDequeue */ static void Main(string[] args) { _ConcurrenProducts = new ConcurrentQueue<Product>(); /*執行添加操作*/ Console.WriteLine("執行添加操作"); Parallel.Invoke(AddConcurrenProducts, AddConcurrenProducts); Console.WriteLine("ConcurrentQueue<Product> 當前數據量為:" + _ConcurrenProducts.Count); /*執行TryPeek操作 嘗試返回不移除*/ Console.WriteLine("執行TryPeek操作 嘗試返回不移除"); Parallel.Invoke(PeekConcurrenProducts, PeekConcurrenProducts); Console.WriteLine("ConcurrentQueue<Product> 當前數據量為:" + _ConcurrenProducts.Count); /*執行TryDequeue操作 嘗試返回並移除*/ Console.WriteLine("執行TryDequeue操作 嘗試返回並移除"); Parallel.Invoke(DequeueConcurrenProducts, DequeueConcurrenProducts); Console.WriteLine("ConcurrentQueue<Product> 當前數據量為:" + _ConcurrenProducts.Count); Console.ReadLine(); } /*執行集合數據添加操作*/ static void AddConcurrenProducts() { Parallel.For(0, 100, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; _ConcurrenProducts.Enqueue(product); }); } /*嘗試返回 但不移除*/ static void PeekConcurrenProducts() { Parallel.For(0, 2, (i) => { Product product = null; bool excute = _ConcurrenProducts.TryPeek(out product); Console.WriteLine(product.Name); }); } /*嘗試返回 並 移除*/ static void DequeueConcurrenProducts() { Parallel.For(0, 2, (i) => { Product product = null; bool excute = _ConcurrenProducts.TryDequeue(out product); Console.WriteLine(product.Name); }); } } class Product { public string Name { get; set; } public string Category { get; set; } public int SellPrice { get; set; } } View Code

需要注意 TryDequeue  和  TryPeek 的無序性,在多線程下

ConcurrentStack  是完全無鎖的,能夠支持並發的添加元素,後進先出。下面貼代碼,詳解見注釋:

private static object o = new object(); /*定義 Stack*/ private static Stack<Product> _Products { get; set; } private static ConcurrentStack<Product> _ConcurrenProducts { get; set; } /* coder:釋迦苦僧 * 代碼中 創建三個並發線程 來操作_Products 和 _ConcurrenProducts 集合,每次添加 30000 條數據 查看 一般Stack 和 多線程安全下的 ConcurrentStack 執行情況 */ static void Main(string[] args) { Thread.Sleep(1000); _Products = new Stack<Product>(); Stopwatch swTask = new Stopwatch(); swTask.Start(); /*創建任務 t1 t1 執行 數據集合添加操作*/ Task t1 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t2 t2 執行 數據集合添加操作*/ Task t2 = Task.Factory.StartNew(() => { AddProducts(); }); /*創建任務 t3 t3 執行 數據集合添加操作*/ Task t3 = Task.Factory.StartNew(() => { AddProducts(); }); Task.WaitAll(t1, t2, t3); swTask.Stop(); Console.WriteLine("List<Product> 當前數據量為:" + _Products.Count); Console.WriteLine("List<Product> 執行時間為:" + swTask.ElapsedMilliseconds); Thread.Sleep(1000); _ConcurrenProducts = new ConcurrentStack<Product>(); Stopwatch swTask1 = new Stopwatch(); swTask1.Start(); /*創建任務 tk1 tk1 執行 數據集合添加操作*/ Task tk1 = Task.Factory.StartNew(() => { AddConcurrenProducts(); }); /*創建任務 tk2 tk2 執行 數據集合添加操作*/ Task tk2 = Task.Factory.StartNew(() => { AddConcurrenProducts(); }); /*創建任務 tk3 tk3 執行 數據集合添加操作*/ Task tk3 = Task.Factory.StartNew(() => { AddConcurrenProducts(); }); Task.WaitAll(tk1, tk2, tk3); swTask1.Stop(); Console.WriteLine("ConcurrentStack<Product> 當前數據量為:" + _ConcurrenProducts.Count); Console.WriteLine("ConcurrentStack<Product> 執行時間為:" + swTask1.ElapsedMilliseconds); Console.ReadLine(); } /*執行集合數據添加操作*/ static void AddProducts() { Parallel.For(0, 30000, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; lock (o) { _Products.Push(product); } }); } /*執行集合數據添加操作*/ static void AddConcurrenProducts() { Parallel.For(0, 30000, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; _ConcurrenProducts.Push(product); }); } } class Product { public string Name { get; set; } public string Category { get; set; } public int SellPrice { get; set; } } View Code

class Program { private static object o = new object(); private static ConcurrentStack<Product> _ConcurrenProducts { get; set; } /* coder:釋迦苦僧 * ConcurrentQueue 下的 TryPeek 和 TryPop */ static void Main(string[] args) { _ConcurrenProducts = new ConcurrentStack<Product>(); /*執行添加操作*/ Console.WriteLine("執行添加操作"); Parallel.Invoke(AddConcurrenProducts, AddConcurrenProducts); Console.WriteLine("ConcurrentStack<Product> 當前數據量為:" + _ConcurrenProducts.Count); /*執行TryPeek操作 嘗試返回不移除*/ Console.WriteLine("執行TryPeek操作 嘗試返回不移除"); Parallel.Invoke(PeekConcurrenProducts, PeekConcurrenProducts); Console.WriteLine("ConcurrentStack<Product> 當前數據量為:" + _ConcurrenProducts.Count); /*執行TryDequeue操作 嘗試返回並移除*/ Console.WriteLine("執行TryPop操作 嘗試返回並移除"); Parallel.Invoke(PopConcurrenProducts, PopConcurrenProducts); Console.WriteLine("ConcurrentStack<Product> 當前數據量為:" + _ConcurrenProducts.Count); Console.ReadLine(); } /*執行集合數據添加操作*/ static void AddConcurrenProducts() { Parallel.For(0, 100, (i) => { Product product = new Product(); product.Name = "name" + i; product.Category = "Category" + i; product.SellPrice = i; _ConcurrenProducts.Push(product); }); } /*嘗試返回 但不移除*/ static void PeekConcurrenProducts() { Parallel.For(0, 2, (i) => { Product product = null; bool excute = _ConcurrenProducts.TryPeek(out product); Console.WriteLine(product.Name); }); } /*嘗試返回 並 移除*/ static void PopConcurrenProducts() { Parallel.For(0, 2, (i) => { Product product = null; bool excute = _ConcurrenProducts.TryPop(out product); Console.WriteLine(product.Name); }); } } class Product { public string Name { get; set; } public string Category { get; set; } public int SellPrice { get; set; } } View Code

對於並發下的其他集合,我這邊就不做代碼案列了,大家可以通過下面的鏈接查看,如有問題,歡迎指正

http://msdn.microsoft.com/zh-cn/library/system.collections.concurrent(v=vs.110).aspx

 

作者:釋迦苦僧 出處:http://www.cnblogs.com/woxpp/p/3935557.html
本文版權歸作者和博客園共有,歡迎轉載,但未經作者同意必須保留此段聲明,且在文章頁面明顯位置給出原文連接。


C語言中 ^怎使用

a1 = 0x01; //0000 0001
a2 = 0x00; //0000 0000
a3 = 0x03; //0000 0011
a4 = 0x02; //0000 0010

b1 = a1 ^ a2; //0000 0001
b2 = a1 ^ a3; //0000 0010
b3 = a1 ^ a4; //0000 0011

^異或運算符,位值相同為0,不同為1,見上示例.

//
簡單實際問題舉例:
======\=======\=======
======a=======b=======
上面是2條電路,2個開關分別為a和b,打開狀態:\[1],關閉狀態:/[0].
若同時打開或者關閉,兩條電路均不通.
若a打開[1],b關閉[0],電路1通電
======\=======/=======
若a關閉[0],b打開[1],電路2通電
======/=======\=======
綜上,電路在a,b狀態相同時不通[0],在a,b不同時通電[1].
 

C語言中 ^怎使用

a1 = 0x01; //0000 0001
a2 = 0x00; //0000 0000
a3 = 0x03; //0000 0011
a4 = 0x02; //0000 0010

b1 = a1 ^ a2; //0000 0001
b2 = a1 ^ a3; //0000 0010
b3 = a1 ^ a4; //0000 0011

^異或運算符,位值相同為0,不同為1,見上示例.

//
簡單實際問題舉例:
======\=======\=======
======a=======b=======
上面是2條電路,2個開關分別為a和b,打開狀態:\[1],關閉狀態:/[0].
若同時打開或者關閉,兩條電路均不通.
若a打開[1],b關閉[0],電路1通電
======\=======/=======
若a關閉[0],b打開[1],電路2通電
======/=======\=======
綜上,電路在a,b狀態相同時不通[0],在a,b不同時通電[1].
 

  1. 上一頁:
  2. 下一頁:
Copyright © 程式師世界 All Rights Reserved