C#——高级同步原语

我就是我 2022-07-17 16:21 201阅读 0赞

## 序 ##

同样是同步的知识点，上一篇只要关注数据同步，这篇将放大视角，看看两个以上的Task或者两个以上的线程应该怎么样协同工作。

### Barrier ###

System.Threading.Barrier primitive的工作原理是这样的：Barrier初始化后，将等待特定数量的信号到来，这个数量在Barrier初始化时指定，在所指定的信号个数已经到来后，Barrier将执行一个指定的动作，这个动作也是在Barrier初始化时指定。Barrier在执行动作过后，将会重置，这时又将等待特定数量的信号到来，再执行指定动作。信号通过成员函数SignalAndWait()来发送，执行SignalAndWait()函数的Task或者线程将会投入等待，Barrier将等待特定数量的信号到达，然后Barrier执行完指定动作后被重置，这时SignalAndWait()函数所在的Task或者线程将继续运行。在程序的运行过程中，可以通过成员函数AddParticipant()和RemoveParticpant()来增加或者减少需要等待的信号数量。例程如下：

static void test0()
            {
                // create the array of bank accounts
                BankAccount[] accounts = new BankAccount[5];
                for (int i = 0; i < accounts.Length; i++)
                {
                    accounts[i] = new BankAccount();
                }
                // create the total balance counter
                int totalBalance = 0;
                // create the barrier
                Barrier barrier = new Barrier(5, (myBarrier) => {
                    // zero the balance
                    totalBalance = 0;
                    // sum the account totals
                    foreach (BankAccount account in accounts)
                    {
                        totalBalance += account.Balance;
                    }
                    // write out the balance
                    Console.WriteLine("Total balance: {0}", totalBalance);
                });
                // define the tasks array
                Task[] tasks = new Task[5];
                // loop to create the tasks
                for (int i = 0; i < tasks.Length; i++)
                {
                    tasks[i] = new Task((stateObj) => {
                        // create a typed reference to the account
                        BankAccount account = (BankAccount)stateObj;
                        // start of phase
                        Random rnd = new Random();
                        for (int j = 0; j < 1000; j++)
                        {
                            //account.Balance += rnd.Next(1, 100);
                            account.Balance += 1;
                        }
                        // end of phase
                        // tell the user that this task has has completed the phase
                        Console.WriteLine("Task {0}, phase {1} ended",
                        Task.CurrentId, barrier.CurrentPhaseNumber);
                        // signal the barrier
                        barrier.SignalAndWait();
                        // start of phase
                        // alter the balance of this Task's account using the total balance
                        // deduct 10% of the difference from the total balance
                        account.Balance -= (totalBalance - account.Balance) / 10;
                        // end of phase
                        // tell the user that this task has has completed the phase
                        Console.WriteLine("Task {0}, phase {1} ended",
                        Task.CurrentId, barrier.CurrentPhaseNumber);
                        // signal the barrier
                        barrier.SignalAndWait();
                    },
                    accounts[i]);
                }
                // start the task
                foreach (Task t in tasks)
                {
                    t.Start();
                }
                // wait for all of the tasks to complete
                Task.WaitAll(tasks);
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }

Barrier设计用于“多阶段的并行算法”中，可以想象一下某算法有几个阶段，每个阶段又可以拆分给几个Task或者线程中同时运行，当某一阶段的所有工作（分配给了几个Task或者线程来完成）都完成后，那么将一起进入下一阶段，而在下一阶段，每个Task或者线程又有各自的工作要做。

Barrier的关键成员函数：

![Center][]  
使用Barrier容易形成死锁，这是因为当某个线程抛出异常时，其他线程将陷入永远的等待，解决办法将在以后的文章中分享。

### CountDownEvent ###

CountDownEvent与Barrier相似，所不同的是，CountDownEvent调用成员函数Wait()将阻塞，直至成员函数Signal() 被调用达特定的次数，这时CountDownEvent称作就绪态，对于处于就绪态的CountDownEvent，调用Wait()函数将不会再阻塞，只有手动调用Reset()函数后，调用Wait()函数将再次阻塞。CountDownEvent可以通过TryAddCount()和AddCount()函数来增加函数Signal() 需被调用的次数，但只有当CountDownEvent处于未就绪态时才会成功。否则根据调用函数的不同，将有可能抛出异常。例程如下：

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace Listing_15
    {
        class Listing_15
        {
            static void Main(string[] args)
            {
                // create a CountDownEvent with a condition
                // counter of 5
                CountdownEvent cdevent = new CountdownEvent(5);
                // create a Random that we will use to generate
                // sleep intervals
                Random rnd = new Random();
                // create 5 tasks, each of which will wait for
                // a random period and then signal the event
                Task[] tasks = new Task[6];
                for (int i = 0; i < tasks.Length; i++)
                {
                    // create the new task
                    tasks[i] = new Task(() => {
                        // put the task to sleep for a random period
                        // up to one second
                        Thread.Sleep(rnd.Next(500, 1000));
                        // signal the event
                        Console.WriteLine("Task {0} signalling event", Task.CurrentId);
                        cdevent.Signal();
                    });
                };
                // create the final task, which will rendezous with the other 5
                // using the count down event
                tasks[5] = new Task(() => {
                    // wait on the event
                    Console.WriteLine("Rendezvous task waiting");
                    cdevent.Wait();
                    Console.WriteLine("Event has been set");
                });
                // start the tasks
                foreach (Task t in tasks)
                {
                    t.Start();
                }
                Task.WaitAll(tasks);
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

CountDownEvent的主要成员如下：

![Center 1][]

### ManualResetEventSlim ###

不同于上面所说的两个同步原语，ManualResetEventSlim只要调用Set()函数，所有的Wait()将不再阻塞，直至手动调用Reset()函数，Wait()将再次阻塞。例程如下：

using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    using System.Threading;
    
    namespace _manualReset
    {
        class Program
        {
            static void Main(string[] args)
            {
                // create the primtive
                ManualResetEventSlim manualResetEvent
                = new ManualResetEventSlim();
                // create the cancellation token source
                CancellationTokenSource tokenSource
                = new CancellationTokenSource();
                // create and start the task that will wait on the event
                Task waitingTask = Task.Factory.StartNew(() => {
                    int i = 0;
                    while (true)
                    {
                        // wait on the primitive
                        manualResetEvent.Wait(tokenSource.Token);
                        // print out a message
                        Console.WriteLine("Waiting task active {0}",i++);
                    }
                }, tokenSource.Token);
                // create and start the signalling task
                Task signallingTask = Task.Factory.StartNew(() => {
                    // create a random generator for sleep periods
                    Random rnd = new Random();
                    // loop while the task has not been cancelled
                    while (!tokenSource.Token.IsCancellationRequested)
                    {
                        // go to sleep for a random period
                        tokenSource.Token.WaitHandle.WaitOne(rnd.Next(500, 2000));
                        // set the event
                        manualResetEvent.Set();
                        Console.WriteLine("Event set");
                        // go to sleep again
                        tokenSource.Token.WaitHandle.WaitOne(rnd.Next(500, 2000));
                        // reset the event
                        manualResetEvent.Reset();
                        Console.WriteLine("Event reset");
                    }
                    // if we reach this point, we know the task has been cancelled
                    tokenSource.Token.ThrowIfCancellationRequested();
                }, tokenSource.Token);
                // ask the user to press return before we cancel
                // the token and bring the tasks to an end
                Console.WriteLine("Press enter to cancel tasks");
                Console.ReadLine();
                // cancel the token source and wait for the tasks
                tokenSource.Cancel();
                try
                {
                    Task.WaitAll(waitingTask, signallingTask);
                }
                catch (AggregateException)
                {
                    // discard exceptions
                }
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

ManualResetEventSlim的主要成员如下：

![Center 2][]

### AutoResetEvent  ###

AutoResetEvent is similar to ManualResetEventSlim，所不同的是，AutoResetEvent将自动重置，并且每次调用Set()都只唤醒一个WaitOne()，而唤醒哪个WaitOne()，这是不确定的。例程如下：

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace Listing_17
    {
        class Listing_17
        {
            static void Main(string[] args)
            {
                // create the primtive
                AutoResetEvent arEvent = new AutoResetEvent(false);
                // create the cancellation token source
                CancellationTokenSource tokenSource
                = new CancellationTokenSource();
                // create and start the task that will wait on the event
                for (int i = 0; i < 3; i++)
                {
                    Task.Factory.StartNew(() =>
                    {
                        while (!tokenSource.Token.IsCancellationRequested)
                        {
                            // wait on the primtive
                            arEvent.WaitOne();
                            // print out a message when we are released
                            Console.WriteLine("Task {0} released", Task.CurrentId);
                        }
                        // if we reach this point, we know the task has been cancelled
                        tokenSource.Token.ThrowIfCancellationRequested();
                    }, tokenSource.Token);
                }
                // create and start the signalling task
                Task signallingTask = Task.Factory.StartNew(() =>
                {
                    // loop while the task has not been cancelled
                    while (!tokenSource.Token.IsCancellationRequested)
                    {
                        // go to sleep for a random period
                        tokenSource.Token.WaitHandle.WaitOne(500);
                        // set the event
                        arEvent.Set();
                        Console.WriteLine("Event set");
                    }
                    // if we reach this point, we know the task has been cancelled
                    tokenSource.Token.ThrowIfCancellationRequested();
                }, tokenSource.Token);
                // ask the user to press return before we cancel
                // the token and bring the tasks to an end
                Console.WriteLine("Press enter to cancel tasks");
                Console.ReadLine();
                // cancel the token source and wait for the tasks
                tokenSource.Cancel();
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

AutoResetEvent的主要成员如下：

![Center 3][]

### SemaphoreSlim ###

这是最后的大BOSS，也即我们在《操作系统》课程中所学到的——信号量。

SemaphoreSlim的主要成员函数是Wait()和Release()，在没有资源可用时，调用Wait()将阻塞；调用Release()将释放零个或者多个被占用的资源，这时，若Wait()得到可以的资源，那么将继续运行，否则将继续阻塞，换句话说，Release()将唤醒零个或者多个被阻塞的线程或者Task。资源的总数会在SemaphoreSlim初始化时指定，并且资源的总数可通过Release()在运行时增加。例程如下：

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace Listing_18
    {
        class Listing_18
        {
            static void Main(string[] args)
            {
                // create the primtive
                SemaphoreSlim semaphore = new SemaphoreSlim(2);
                // create the cancellation token source
                CancellationTokenSource tokenSource
                = new CancellationTokenSource();
                // create and start the task that will wait on the event
                for (int i = 0; i < 10; i++)
                {
                    Task.Factory.StartNew(() =>
                    {
                        while (true)
                        {
                            semaphore.Wait(tokenSource.Token);
                            // print out a message when we are released
                            Console.WriteLine("Task {0} released", Task.CurrentId);
                        }
                    }, tokenSource.Token);
                }
                // create and start the signalling task
                Task signallingTask = Task.Factory.StartNew(() =>
                {
                    // loop while the task has not been cancelled
                    while (!tokenSource.Token.IsCancellationRequested)
                    {
                        // go to sleep for a random period
                        tokenSource.Token.WaitHandle.WaitOne(500);
                        // signal the semaphore
                        semaphore.Release(2);
                        Console.WriteLine("Semaphore released");
                    }
                    // if we reach this point, we know the task has been cancelled
                    tokenSource.Token.ThrowIfCancellationRequested();
                }, tokenSource.Token);
                // ask the user to press return before we cancel
                // the token and bring the tasks to an end
                Console.WriteLine("Press enter to cancel tasks");
                Console.ReadLine();
                // cancel the token source and wait for the tasks
                tokenSource.Cancel();
                // wait for input before exiting
                Console.WriteLine("Press enter to finish");
                Console.ReadLine();
            }
        }
    }

SemaphoreSlim的主要成员如下：

![Center 4][]

[Center]: /images/20220717/e52d29c7b33d497ab55f71d741dd6baa.png
[Center 1]: /images/20220717/cd3887c4d30c421cb6b8a08d0375058e.png
[Center 2]: /images/20220717/2affa7669c2a4d3e98f21d13e7d7452e.png
[Center 3]: /images/20220717/2dc341b117f94f798555f3302883860a.png
[Center 4]: /images/20220717/16dd3dbfe3d04f02925eec859fe39053.png