碼上快樂

相關內容    簡體    繁體

C#並行編程-並發集合

本文轉載自   查看原文   2014-08-27 14:52   10838    並發集合/ ConcurrentQueue/ .NET/ ConcurrentStack/ 並行編程

菜鳥學習並行編程，參考《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在一定程度上也帶來了損耗，如下圖所示：

ConcurrentQueue 還有另外兩種方法：TryDequeue  嘗試移除並返回 和 TryPeek 嘗試返回但不移除，下面貼代碼：

    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

ConcurrentStack 還有另外兩種方法：TryPop 嘗試移除並返回 和 TryPeek 嘗試返回但不移除，下面貼代碼：

    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
本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接。


免責聲明！

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



猜您在找 《C#並行編程高級教程》第4章 並發集合 筆記 C#並發編程之初識並行編程 C#多線程開發：並行、並發與異步編程 談談C#多線程開發：並行、並發與異步編程 5天玩轉C#並行和多線程編程 —— 第二天 並行集合和PLinq C#並發集合 C#集合之並發集合 C#並行編程--並行任務 C#並行編程-Task C#並行編程-Parallel
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM