C#多线程编程实战（二）：线程同步

2.1 简介

竞争条件：多个线程同时使用共享对象。需要同步这些线程使得共享对象的操作能够以正确的顺序执行

线程同步问题：多线程的执行并没有正确的同步，当一个线程执行递增和递减操作时，其他线程需要依次等待

线程同步解决方案：

无须共享对象：大部分时候可以通过重新设计来移除共享对象，去掉复杂的同步构造，避免多线程使用单一对象

必须共享对象：只使用原子操作，一个操作只占用一个量子的时间，无须实现其他线程等待当前操作完成

内核模式：将等待的线程置于阻塞状态，消耗少量的CPU资源，但会引入至少一次上下文切换，适用于线程等待较长时间

用户模式：只是简单的等待，线程等待会浪费CPU时间但是可以节省上下文切换消耗的CPU时间，适用于线程等待较短时间

混合模式：先尝试用户模式，如果等待时间较长，则会切换到内核模式

2.2 执行基本的原子操作

using System;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

        {

            Console.WriteLine("Incorrect counter");

            //没有限定，会遇到竞争条件，得出的结果大部分不是正确的

            var c = new Counter();

            var t1 = new Thread(() => TestCounter(c));

            var t2 = new Thread(() => TestCounter(c));

            var t3 = new Thread(() => TestCounter(c));

            t1.Start();

            t2.Start();

            t3.Start();

            t1.Join();

            t2.Join();

            t3.Join();

            Console.WriteLine($"Total count:{c.Count}");

            Console.WriteLine("--------------------------");

            Console.WriteLine("Correct counter");

            //使用Interlocked类提供的原子操作方法，无需锁定任何对象可得出正确结果

            var c1 = new CounterNoLock();

            t1 = new Thread(() => TestCounter(c1));

            t2 = new Thread(() => TestCounter(c1));

            t3 = new Thread(() => TestCounter(c1));

            t1.Start();

            t2.Start();

            t3.Start();

            t1.Join();

            t2.Join();

            t3.Join();

            Console.WriteLine($"Total count:{c1.Count}");

            Console.ReadLine();

        }

        static void TestCounter(CounterBase c)

        {

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

            {

                c.Increment();

                c.Decrement();

            }

        }

        class Counter : CounterBase

        {

            private int _count;

            public int Count { get { return _count; } }

            public override void Decrement()

            {

                _count--;

            }

            public override void Increment()

            {

                _count++;

            }

        }

        class CounterNoLock : CounterBase

        {

            private int _count;

            public int Count { get { return _count; } }

            public override void Decrement()

            {

                //Interlocked提供了Increment()、Decrement()和Add等基本数学操作的原子方法

                Interlocked.Decrement(ref _count);

            }

            public override void Increment()

            {

                Interlocked.Increment(ref _count);

            }

        }

        abstract class CounterBase

        {

            public abstract void Increment();

            public abstract void Decrement();

        }

    }

}

注释：Interlocked提供了Increment()、Decrement()和Add等基本数学操作的原子方法，不用锁也可以得出正确结果

2.3 使用Mutex类

using System;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

        {

            const string MutexName = "CSharpThreadingCookbook";

            //Mutex是一种原始的同步方式，只对一个线程授予对共享资源的独占访问

            //定义一个指定名称的互斥量，设置initialOwner标志为false

            using (var m = new Mutex(false, MutexName))

            {

                //如果互斥量已经被创建，获取互斥量，否则就执行else语句

                if (!m.WaitOne(TimeSpan.FromSeconds(), false))

                {

                    Console.WriteLine("Second instance is running!");

                }

                else

                {

                    Console.WriteLine("Running!");

                    Console.ReadLine();

                    m.ReleaseMutex();

                }

            }

            //如果再运行同样的程序，则会在5秒内尝试获取互斥量，如果第一个程序按下了任何键，第二个程序开始执行。

            //如果等待5秒钟，第二个程序将无法获取该互斥量

        }

    }

}

注释：互斥量是全局操作对象，必须正确关闭，最好用using

2.4 使用SemaphoreSlim类

using System;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

         {

            //启动6个线程，启动的顺序不一样

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

            {

                string threadName = "Thread " + i;

                int secondsToWait =  +  * i;

                var t = new Thread(() => AccessDatabase(threadName, secondsToWait));

                t.Start();

            }

            Console.ReadLine();

        }

        //SemaphoreSlim的构造函数参数为允许的并发线程数目

        static SemaphoreSlim semaphore = new SemaphoreSlim();

        static void AccessDatabase(string name, int seconds)

        {

            Console.WriteLine($"{name} waits to access a database");

            semaphore.Wait();

            Console.WriteLine($"{name} was granted an access to a database");

            Thread.Sleep(TimeSpan.FromSeconds(seconds));

            Console.WriteLine($"{name} is Completed");

            //调用Release方法说明线程已经完成，可以开启一个新的线程了

            semaphore.Release();

        }

    }

}

注释：这里使用了混合模式，允许我们在等待时间很短的情况下无需上下文切换。SemaphoreSlim并不使用Windows内核信号量，而且也不支持进程间同步

2.5 使用AutoResetEvent类

using System;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

         {

            Thread t = new Thread(() => Process());

            t.Start();

            Console.WriteLine("Waiting for another thread to complete work");

            //开启一个线程后workEvent等待，直到收到Set信号

            workEvent.WaitOne();

            Console.WriteLine("First operation is complete");

            Console.WriteLine("Performing an operation on a main thread");

            Thread.Sleep(TimeSpan.FromSeconds());

            mainEvent.Set();

            Console.WriteLine("Now running the second operation on a second thread");

            workEvent.WaitOne();

            Console.WriteLine("Second operation is complete");

            Console.ReadLine();

        }

        //初始状态为unsignaled，子线程向主线程发信号

        private static AutoResetEvent workEvent = new AutoResetEvent(false);

        //初始状态为unsignaled，主线程向子线程发信号

        private static AutoResetEvent mainEvent = new AutoResetEvent(false);

        static void Process(int seconds)

        {

            Console.WriteLine("Starting a long running work...");

            Thread.Sleep(TimeSpan.FromSeconds(seconds));

            Console.WriteLine("Work is done!");

            workEvent.Set();//将事件设为终止状态允许一个或多个线程继续

            Console.WriteLine("Waiting for a main thread to complete its work");

            mainEvent.WaitOne();//阻止当前线程，直到mainEvent收到信号

            Console.WriteLine("Starting second operation...");

            Thread.Sleep(TimeSpan.FromSeconds(seconds));

            Console.WriteLine("Work is done");

            workEvent.Set();

        }

    }

}

注释：AutoResetEvent采用的是内核模式，所以等待时间不能太长

2.6 使用ManualResetEventSlim类

using System;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

         {

            Thread t1 = new Thread(() => TravelThroughGates("Thread 1", ));

            Thread t2 = new Thread(() => TravelThroughGates("Thread 2", ));

            Thread t3 = new Thread(() => TravelThroughGates("Thread 3", ));

            t1.Start();

            t2.Start();

            t3.Start();

            Thread.Sleep(TimeSpan.FromSeconds());

            Console.WriteLine("The gates are now open");

            mainEvent.Set();//将时间设置为有信号，从而让一个或多个等待该事件的线程继续

            Thread.Sleep(TimeSpan.FromSeconds());

            mainEvent.Reset();//将事件设置为非终止，从而导致线程受阻

            Console.WriteLine("The gates have been closed!");

            Thread.Sleep(TimeSpan.FromSeconds());

            Console.WriteLine("The gates are now open for the second time");

            mainEvent.Set();

            Thread.Sleep(TimeSpan.FromSeconds());

            Console.WriteLine("The gates have been closed!");

            mainEvent.Reset();

            Console.ReadLine();

        }

        static ManualResetEventSlim mainEvent = new ManualResetEventSlim(false);

        static void TravelThroughGates(string threadName, int seconds)

        {

            Console.WriteLine($"{threadName} falls to sleep");

            Thread.Sleep(TimeSpan.FromSeconds(seconds));

            Console.WriteLine($"{threadName} waits for the gates to open!");

            mainEvent.Wait();//阻止当前线程

            Console.WriteLine($"{threadName} enter the gates!");

        }

    }

}

注释：ManualResetEventSlim工作方式像人群通过的大门，一直保持大门敞开直到调用reset，set相当于打开大门，reset相当于关闭大门

2.7 使用CountDownEvent类

using System;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

         {

            Console.WriteLine("Starting two operations");

            Thread t1 = new Thread(() => PerformOperation("Operation 1 is completed", ));

            Thread t2 = new Thread(() => PerformOperation("Operation 2 is completed", ));

            t1.Start();

            t2.Start();

            //开启了两个线程，调用Wait方法阻止当前线程，知道所有线程都完成

            countdown.Wait();

            Console.WriteLine("Both operations have been completed");

            countdown.Dispose();

            Console.ReadLine();

        }

        //计数器初始化CountdownEvent实例，计数器表示：当计数器个数完成操作发出信号

        static CountdownEvent countdown = new CountdownEvent();

        static void PerformOperation(string message, int seconds)

        {

            Thread.Sleep(TimeSpan.FromSeconds(seconds));

            Console.WriteLine(message);

            //向CountdownEvent注册信息，并减少当前计数器数值

            countdown.Signal();

        }

    }

}

注释：如果Signal方法没有达到指定的次数，那么countdown.wait()会一直等待，所以请确保所有线程完成后都要调用Signal方法

2.8 使用Barrier类

using System;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

         {

            Thread t1 = new Thread(() => PlayMusic("the gutarist", "play an amazing solo", ));

            Thread t2 = new Thread(() => PlayMusic("the signer", "sing his song", ));

            t1.Start();

            t2.Start();

            Console.ReadLine();

        }

        //后面的Lamda表达式是回调函数。执行完SignalAndWait后执行

        static Barrier barrier = new Barrier(, b=>Console.WriteLine($"End of phase {b.CurrentPhaseNumber + 1}"));

        static void PlayMusic(string name, string message, int seconds)

        {

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

            {

                Console.WriteLine("===========================");

                Thread.Sleep(TimeSpan.FromSeconds(seconds));

                Console.WriteLine($"{name} starts to {message}");

                Thread.Sleep(TimeSpan.FromSeconds(seconds));

                Console.WriteLine($"{name} finishes to {message}");

                //等所有调用线程都结束

                barrier.SignalAndWait();

            }

        }

    }

}

注释：

2.9 使用ReaderWriterlockSlim类

using System;

using System.Collections.Generic;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

         {

            new Thread(Read) { IsBackground = true }.Start();

            new Thread(Read) { IsBackground = true }.Start();

            new Thread(Read) { IsBackground = true }.Start();

            new Thread(() => Write("Thread 1")) { IsBackground = true }.Start();

            new Thread(() => Write("Thread 2")) { IsBackground = true }.Start();

            Thread.Sleep(TimeSpan.FromSeconds());

            Console.ReadLine();

        }

        //实现线程安全

        static ReaderWriterLockSlim _rw = new ReaderWriterLockSlim();

        static Dictionary<int, int> items = new Dictionary<int, int>();

        static void Read()

        {

            Console.WriteLine("Reading contents of a dictionary");

            while (true)

            {

                try

                {

                    //读锁

                    _rw.EnterReadLock();

                    foreach (var key in items.Keys)

                    {

                        Thread.Sleep(TimeSpan.FromSeconds(0.1));

                    }

                }

                finally

                {

                    //计数为0时退出读取模式

                    _rw.ExitReadLock();

                }

            }

        }

        static void Write(string threadName)

        {

            while (true)

            {

                try

                {

                    int newKey = new Random().Next();

                    _rw.EnterUpgradeableReadLock();

                    if (!items.ContainsKey(newKey))

                    {

                        try

                        {

                            //写锁

                            _rw.EnterWriteLock();

                            items[newKey] = ;

                            Console.WriteLine($"New key {newKey} is added to a dictionary by a {threadName}");

                        }

                        finally

                        {

                            //计数为0时退出写入模式

                            _rw.ExitWriteLock();

                        }

                    }

                    Thread.Sleep(TimeSpan.FromSeconds(0.1));

                }

                finally

                {

                    //计数为0时退出可升级模式

                    _rw.ExitUpgradeableReadLock();

                }

            }

        }

    }

}

注释：从集合读取数据时，根据当前数据决定是否获取一个写锁并修改该集合。获取写锁后集合会处于阻塞状态。

2.10 使用SpinWait类

using System;

using System.Collections.Generic;

using System.Threading;

namespace MulityThreadNote

{

    class Program

    {

        static void Main(string[] args)

         {

            Thread t1 = new Thread(UserModeWait);

            Thread t2 = new Thread(HybridSpinWait);

            Console.WriteLine("Running user mode waiting");

            t1.Start();

            Thread.Sleep();

            _isComplete = true;

            Thread.Sleep(TimeSpan.FromSeconds());

            _isComplete = false;

            Console.WriteLine("Running hybrid SpinWait construct waiting");

            t2.Start();

            Thread.Sleep();

            _isComplete = true;

            Console.ReadLine();

        }

        //volatile 一个字段可能会被多个线程同时修改，不会被编译器和处理器优化为只能被单个线程访问

        static volatile bool _isComplete = false;

        static void UserModeWait()

        {

            while (!_isComplete)

            {

                Console.WriteLine(".");

            }

            Console.WriteLine();

            Console.WriteLine("Waiting is complete");

        }

        static void HybridSpinWait()

        {

            var w = new SpinWait();

            while (!_isComplete)

            {

                //执行单一自旋

                w.SpinOnce();

                //NextSpinWillYield:获取对SpinOnce的下一次调用是否将产生处理，同时触发强制上下文切换

                //显示线程是否切换为阻塞状态

                Console.WriteLine(w.NextSpinWillYield);

            }

            Console.WriteLine("Waiting is complete");

        }

    }

}

注释：

巴特西