C#高级编程五十八天----并行集合

并行集合

对于并行任务,与其相关紧密的就是对一些共享资源,数据结构的并行訪问.常常要做的就是对一些队列进行加锁-解锁,然后运行类似插入,删除等等相互排斥操作. .NET4提供了一些封装好的支持并行操作数据容器,能够降低并行编程的复杂程度.

 

并行集合的命名空间:System.Collections.Concurrent

并行容器:

ConcurrentQueue

ConcurrentStack

ConcurrentBag: 一个无序的数据结构集,当不考虑顺序时很实用.

BlockingCollection:与经典的堵塞队列数据结构类似

ConcurrentDictoinary

 

以上这些集合在某种程度上使用了无锁技术(CAS和内存屏蔽),与加相互排斥锁相比获得了性能的提升.可是在串行程序中,最好不要使用这些集合,他们必定会影响性能.

 

ConcurrentQueue使用方法与实例

其全然无锁,但当CAS面临资源竞争失败时可能会陷入自旋并重试操作.

Enqueue:在队尾插入元素

TryDequeue:尝试删除对头元素,并通过out參数返回

TryPeek:尝试将对头元素通过out參数返回,但不删除元素

案例:

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using System.Collections.Concurrent;

namespace 并行结合Queue

{

    class Program

    {

        internal static ConcurrentQueue<int> _TestQueue;

        class ThreadWork1 //生产者

        {

            public ThreadWork1()

            { }

            public void run()

            {

                Console.WriteLine("ThreadWork1 run { ");

                for (int i = 0; i < 100; i++)

                {

                    Console.WriteLine("ThreadWork1 producer: "+i);

                    _TestQueue.Enqueue(i);

                }

                Console.WriteLine("ThreadWork1 run } ");

            }

        }

        class ThreadWork2 //consumer

        {

            public ThreadWork2()

            { }

            public void run()

            {

                int i = 0;

                bool IsDequeue = false;

                Console.WriteLine("ThreadWork2 run { ");

                for (; ; )

                {

                    IsDequeue = _TestQueue.TryDequeue(out i);

                    if (IsDequeue)

                    {

                        System.Console.WriteLine("ThreadWork2 consumer: " + i * i + "   ====="); 

                    }

                    if (i==99)

                    {

                        break;

                    }                    

                }

                Console.WriteLine("ThreadWork2 run } ");

            }

        }

        static void StartT1()

        {

            ThreadWork1 work1 = new ThreadWork1();

            work1.run();

        }

 

        static void StartT2()

        {

            ThreadWork2 work2 = new ThreadWork2();

            work2.run();

        }  

        static void Main(string[] args)

        {

            Task t1 = new Task(() => StartT1());

            Task t2 = new Task(() => StartT2());

 

            _TestQueue = new ConcurrentQueue<int>();

 

            Console.WriteLine("Sample 3-1 Main {");

 

            Console.WriteLine("Main t1 t2 started {");

            t1.Start();

            t2.Start();

            Console.WriteLine("Main t1 t2 started }");

 

            Console.WriteLine("Main wait t1 t2 end {");

            Task.WaitAll(t1, t2);

            Console.WriteLine("Main wait t1 t2 end }");

 

            Console.WriteLine("Sample 3-1 Main }");

 

            Console.ReadKey();  

        }

    }

}

 

ConcurrentStact

其全然无锁,但当CAS面临资源竞争失败时可能会陷入自旋并重试操作.

Push:向栈顶插入元素

TryPop:从栈顶弹出元素,而且通过out參数返回

TryPeek:返回栈顶元素,但不弹出

 

案例:

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using System.Collections.Concurrent;

namespace 并行结合Queue

{

    class Program

    {

        internal static ConcurrentStack<int> _TestStack;

        class ThreadWork1 //生产者

        {

            public ThreadWork1()

            { }

            public void run()

            {

                Console.WriteLine("ThreadWork1 run { ");

                for (int i = 0; i < 100; i++)

                {

                    Console.WriteLine("ThreadWork1 producer: "+i);

                    _TestStack.Push(i);

                }

                Console.WriteLine("ThreadWork1 run } ");

            }

        }

        class ThreadWork2 //consumer

        {

            public ThreadWork2()

            { }

            public void run()

            {

                int i = 0;

                bool IsDequeue = false;

                Console.WriteLine("ThreadWork2 run { ");

                for (; ; )

                {

                    IsDequeue = _TestStack.TryPop(out i);

                    if (IsDequeue)

                    {

                        System.Console.WriteLine("ThreadWork2 consumer: " + i * i + "   =====" + i); 

                    }

                    if (i==99)

                    {

                        break;

                    }                    

                }

                Console.WriteLine("ThreadWork2 run } ");

            }

        }

        static void StartT1()

        {

            ThreadWork1 work1 = new ThreadWork1();

            work1.run();

        }

 

        static void StartT2()

        {

            ThreadWork2 work2 = new ThreadWork2();

            work2.run();

        }  

        static void Main(string[] args)

        {

            Task t1 = new Task(() => StartT1());

            Task t2 = new Task(() => StartT2());

 

            _TestStack = new ConcurrentStack<int>();

 

            Console.WriteLine("Sample 4-1 Main {");

 

            Console.WriteLine("Main t1 t2 started {");

            t1.Start();

            t2.Start();

            Console.WriteLine("Main t1 t2 started }");

 

            Console.WriteLine("Main wait t1 t2 end {");

            Task.WaitAll(t1, t2);

            Console.WriteLine("Main wait t1 t2 end }");

 

            Console.WriteLine("Sample 4-1 Main }");

 

            Console.ReadKey();  

        }

    }

}

 

 

ConcurrentBag

一个无序的集合,程序能够向当中插入元素,或删除元素.

在同一个线程中向集合插入,删除元素效率非常高.

Add:向集合中插入元素

TryTake:从集合中取出元素并删除

TryPeek:从集合中取出元素,但不删除元素

 

案例:

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

 

namespace 并行集合ConcurrentBag

{

    class Program

    {

        internal static ConcurrentBag<int> _TestBag;

        class ThreadWork1 //producer

        {

            public ThreadWork1()

            { }

            public void run()

            {

                Console.WriteLine("Threadwork1 run { ");

                for (int i = 0; i < 100; i++)

                {

                    Console.WriteLine("ThreadWork1 producer: "+i);

                    _TestBag.Add(i);

                }

                Console.WriteLine("ThreadWork1 run } ");

            }

        }

        class ThreadWork2//consumer

        {

            public ThreadWork2()

            { }

            public void run()

            {

                bool IsDequeue = false;

                Console.WriteLine("ThreadWork2 run { ");

                for (int i = 0; i < 100; i++)

                {

                    IsDequeue = _TestBag.TryTake(out i);

                    if (IsDequeue)

                    {

                        Console.WriteLine("ThreadWork2 consumer: " + i * i + "======" + i);

                    }

                }

                Console.WriteLine("ThreadWork2 run } ");

            }

        }

        static void Start1()

        {

            ThreadWork1 work1 = new ThreadWork1();

            work1.run();

        }

        static void Start2()

        {

            ThreadWork2 work2 = new ThreadWork2();

            work2.run();

        }

        static void Main(string[] args)

        {

            Task t1 = new Task(() => Start1());

            Task t2 = new Task(() => Start2());

            _TestBag = new ConcurrentBag<int>();

            t1.Start();

            t2.Start();

            Console.WriteLine("Main t1 t2 started }");

 

            Console.WriteLine("Main wait t1 t2 end {");

            Task.WaitAll(t1, t2);

            Console.WriteLine("Main wait t1 t2 end }");

 

            Console.WriteLine("Sample 4-3 Main }");

 

            Console.ReadKey();  

        }

    }

}

 

 

 

BlockingCollection

一个支持界限和堵塞的容器

Add:向容器中插入元素

TryTake:从容器中取出元素并删除

TryPeek:从容器中取出元素,但不删除

CompleteAdding:告诉容器,加入元素完毕.此时假设还想继续加入会发生异常.

IsCompleted:告诉消费者线程,产生者线程还在继续执行中,任务还未完毕.

 

案例:

程序中,消费者线程全然使用While(!__testCollection.IsCompleted)作为退出执行的推断条件.在Work1中,有两条语句被凝视了,当i为50时设置为CompleteAdding,但当继续向当中插入元素时,系统抛出异常,提示无法再继续插入.

案例:

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

 

namespace 并行结合Collection

{

    class Program

    {

        internal static BlockingCollection<int> _testBCollection;

        class ThreadWork1 //producer

        {

            public ThreadWork1()

            { }

            public void run()

            {

                Console.WriteLine("ThreadWork1 run { ");

                for (int i = 0; i < 100; i++)

                {

                    Console.WriteLine("ThreadWork1 producer: "+i);

                    _testBCollection.Add(i);

                    //当i为50时设置为CompleteAdding,但当继续向当中插入元素时,系统抛出异常,提示无法再继续插入.

                    //if (i==50)

                    //{

                    //    _testBCollection.CompleteAdding();

                    //}

                }

                _testBCollection.CompleteAdding();

                Console.WriteLine("ThreadWork1 run } ");

            }

        }

        class ThreadWork2//consumer

        {

            public ThreadWork2()

            { }

            public void run()

            {

                int i = 0;

                

                bool IsDequeue = false;

                Console.WriteLine("ThreadWork2 run { ");

                while (!_testBCollection.IsCompleted)

                {

                    //不明确这里为啥i会自己主动的++

                    //Console.WriteLine("i=================="+i);

                    IsDequeue = _testBCollection.TryTake(out i);

                    if (IsDequeue)

                    {

                        Console.WriteLine("ThreadWork2 consumer: "+i*i+"====="+i);

                       //i++;这句话不加还是会出现自己主动++的操作

                    }

                }

            }

        }

        static void StartT1()

        {

            ThreadWork1 work1 = new ThreadWork1();

            work1.run();

        }

 

        static void StartT2()

        {

            ThreadWork2 work2 = new ThreadWork2();

            work2.run();

        }  

        static void Main(string[] args)

        {

 

            Task t1 = new Task(() => StartT1());

            Task t2 = new Task(() => StartT2());

 

            _testBCollection = new BlockingCollection<int>();

 

            Console.WriteLine("Sample 4-4 Main {");

 

            Console.WriteLine("Main t1 t2 started {");

            t1.Start();

            t2.Start();

            Console.WriteLine("Main t1 t2 started }");

 

            Console.WriteLine("Main wait t1 t2 end {");

            Task.WaitAll(t1, t2);

            Console.WriteLine("Main wait t1 t2 end }");

 

            Console.WriteLine("Sample 4-4 Main }");

 

            Console.ReadKey();  

        }

    }

}

 

分析://_testBCollection.CompleteAdding();//这句话凝视掉work2陷入循环,无法退出

 

ConcurrentDictionary

对于读操作是全然无锁的,当非常多线程要改动数据时,它会使用细粒度的锁.

AddOrUpdate:假设键不存在,方法会在容器中加入新的键和值,假设存在,则更新现有的键和值

GetOrAdd:假设键不存在,方法会向容器中加入新的键和值,假设存在则返回现有的值,并不加入新值.

TryAdd:尝试在容器中加入新的键和值

TryGetValue:尝试依据指定的键获得值

TryUpdate:有条件的更新当前键所相应的值

TryRemove:尝试删除指定的键.

Getenumerator:返回一个可以遍历整个容器的枚举器.

案例:

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

 

namespace 并行集合Dictionary

{

    class Program

    {

        internal static ConcurrentDictionary<int, int> _TestDictionary;

        class ThreadWork1//producer

        {

            public ThreadWork1()

            { }

            public void run()

            {

                System.Console.WriteLine("ThreadWork1 run { ");

                for (int i = 0; i < 100; i++)

                {

                    System.Console.WriteLine("ThreadWork1 producer: " + i);

                    _TestDictionary.TryAdd(i, i);

                }

                System.Console.WriteLine("ThreadWork1 run } ");

            }

        }

        class ThreadWork2//consumer

        {

            public ThreadWork2()

            { }

            public void run()

            {

                bool IsOk = false;

                Console.WriteLine("ThreadWork2 run { ");

                int nCnt = 0,i=0,nValue=0;

                while (nCnt<100)

                {

                    IsOk = _TestDictionary.TryGetValue(i,out nValue);

                    if (IsOk)

                    {

                        Console.WriteLine("ThreadWork2 consumer: "+i*i+"====="+i);

                        nValue *= nValue;

                        nCnt++;

                        i++;

                    }

                }

                Console.WriteLine("ThreadWork2 run } ");

            }

 

        }

        static void StartT1()

        {

            ThreadWork1 work1 = new ThreadWork1();

            work1.run();

        }

 

        static void StartT2()

        {

            ThreadWork2 work2 = new ThreadWork2();

            work2.run();

        }  

        static void Main(string[] args)

        {

            Task t1 = new Task(() => StartT1());

            Task t2 = new Task(() => StartT2());

            bool bIsNext = true;

            int nValue = 0;

            _TestDictionary = new ConcurrentDictionary<int, int>();

            Console.WriteLine("Sample 4-5 Main {");

 

            Console.WriteLine("Main t1 t2 started {");

            t1.Start();

            t2.Start();

            Console.WriteLine("Main t1 t2 started }");

 

            Console.WriteLine("Main wait t1 t2 end {");

            Task.WaitAll(t1, t2);

            Console.WriteLine("Main wait t1 t2 end }");

 

            foreach (var pair in _TestDictionary)

            {

                Console.WriteLine(pair.Key + " : " + pair.Value);

            }

            IEnumerator<KeyValuePair<int, int>> enumer = _TestDictionary.GetEnumerator();

 

            while (bIsNext)

            {

                bIsNext = enumer.MoveNext();

                Console.WriteLine("Key: " + enumer.Current.Key +"  Value: " + enumer.Current.Value);

                _TestDictionary.TryRemove(enumer.Current.Key, out nValue);  

            }

            Console.WriteLine("\n\nDictionary Count: " + _TestDictionary.Count);

 

            Console.WriteLine("Sample 4-5 Main }");

 

            Console.ReadKey();  

        }

    }

}

 

总结说明: .NET4包括的新命名空间System.Collection,Concurrent有几个线程安全的集合类.线程安全的集合可防止多个线程以相互冲突的方式訪问集合.

 

为了对集合进行线程安全的訪问,定义了IPriducerConsumerCollection<T>接口.这个接口中最重要的方法是TryAdd()和TryTake().TryAdd()方法尝试给集合加入一项,但假设集合禁止加入项,这个操作可能失败.为了给出相关信息,TryAdd()方法返回一个布尔值,以说明操作是成功还是失败.TryTake()方法也以这样的方式工作,以通过调用者操作是成功还是失败,并在操作成功时返回集合中的项.