多核时代 .NET Framework 4 中的并行编程8---任务的同步

在并行编程过程中,多个任务同时执行时,就会涉及到任务的同步问题..Net为我们提供很多解决任务同步的类和方法.下面在具体介绍.当然,这些类和方法也适用于处理多线程(Thread)编程的同步问题.

1. Barrier

Barrier类是.Net4中新增加一个类, 它使多个任务能够采用并行方式依据某种算法在多个阶段中协同工作。例如:

BankAccount[] accounts = new BankAccount[5];

for (int i = 0; i < accounts.Length; i++)

{

accounts[i] = new BankAccount();

}

int totalBalance = 0;

System.Threading.Barrier barrier = new System.Threading.Barrier(5, (myBarrier) =>

{

totalBalance = 0;

foreach (var account in accounts)

{

totalBalance += account.Balance;

}

Console.WriteLine("Total balance:{0}", totalBalance);

});

Task[] tasks = new Task[5];

for (int i = 0; i < tasks.Length; i++)

{

tasks[i] = new Task((indata) =>

{

BankAccount account = (BankAccount)indata;

Random rnd = new Random();

for (int j = 0; j < 1000; j++)

{

account.Balance += rnd.Next(1, 100);

}

Console.WriteLine("Task {0},phase {1} ended", Task.CurrentId, barrier.CurrentPhaseNumber);

barrier.SignalAndWait();

account.Balance -= (totalBalance - account.Balance) / 10;

Console.WriteLine("Task {0},phase {1} ended", Task.CurrentId, barrier.CurrentPhaseNumber);

barrier.SignalAndWait();

}, accounts[i]);

}

foreach (var t in tasks)

{

t.Start();

}

Task.WaitAll(tasks);

Console.WriteLine("Press enter to finish");

Console.ReadLine();

通过上面的代码,我们可以知道:

初始化一个Barrier实例,我们可以使用Barrier(Int32, Action<Barrier>)来实例,它的第一个参数表示参数同步线程的数量(也就是将有多少个线程参与进来), Action则表示所有参与者线程到达一个阶段的屏障之后,就会执行Action委托中的代码.如果没有Action委托,则可以传递null,此时就是Barrier(Int32)构造函数了.当参与的线程到达某个阶段时就可以调用Barrier的SignalAndWait();方法, 发出参与者已达到 Barrier 的信号，并等待所有其他参与者也达到屏障.

通俗的讲,Barrier的原理就好比110米栏,刘翔和罗伯斯比赛,当刘翔先跨过第1个栏时就会调用SignalAndWait()方法,告诉大家我已经过了第1个拦,然后等待罗伯斯也跨过第1个栏,当所有人都过了第1个栏,就会调用初始化Barrier时的是Action委托.接着,后续都是这么一个过程.在比赛过程中,如果有其他人加入,则可以调用AddParticipant()方法,如果中途有人退出比赛了,则可以调用RemoveParticpant()退出.

2. CountdownEvent

表示在计数变为零时处于有信号状态的同步基元。其原理是通过一个计数,来监视当前参与者的数量,当某个任务完成后,可以调用CountdownEvent的 Signal()方法表示发出信号,通知任务完成并减少计数.此外,可以调用CountdownEvent 的Wait()方法则阻塞线程直到设置了 CountdownEvent 为止(就是计数为0,也就是所有的参与者都完成了).代码如下:

CountdownEvent cd = new CountdownEvent(5);

Random rnd = new Random();

Task[] tasks = new Task[6];

for (int i = 0; i < tasks.Length; i++)

{

tasks[i] = new Task(() =>

{

Thread.Sleep(rnd.Next(500, 1000));

Console.WriteLine("Task {0} signalling event", Task.CurrentId);

cd.Signal();

});

}

tasks[5] = new Task(() =>

{

cd.Wait();

Console.WriteLine("Event has been set");

});

foreach (Task t in tasks)

{

t.Start();

}

Task.WaitAll(tasks);

Console.WriteLine("Press enter to finish");

Console.ReadLine();

3. ManualResetEventSlim

它是 ManualResetEvent 的简化版本,轻量级的实现. 其Reset方法将事件状态设置为非终止状态，从而导致线程受阻。 其Set 方法将事件状态设置为有信号，从而允许一个或多个等待该事件的线程继续。 其 Wait()方法 阻止当前线程，直到设置了当前 ManualResetEventSlim 为止。代码如下：

ManualResetEventSlim manualResetEvent = new ManualResetEventSlim();

CancellationTokenSource tokenSource = new CancellationTokenSource();

Task waitingTask = Task.Factory.StartNew(() =>

{

while (true)

{

manualResetEvent.Wait(tokenSource.Token);

Console.WriteLine("Waiting task active");

}

}, tokenSource.Token);

Task signallingTask = Task.Factory.StartNew(() =>

{

Random rnd = new Random();

while (!tokenSource.Token.IsCancellationRequested)

{

tokenSource.Token.WaitHandle.WaitOne(rnd.Next(500, 2000));

manualResetEvent.Set();

Console.WriteLine("Event Set");

tokenSource.Token.WaitHandle.WaitOne(rnd.Next(500, 2000));

manualResetEvent.Reset();

Console.WriteLine("Event ReSet");

}

});

Console.WriteLine("Press enter to cancel tasks");

Console.ReadLine();

tokenSource.Cancel();

try

{

Task.WaitAll(waitingTask, signallingTask);

}

catch (AggregateException ex)

{

ex.Flatten().Handle((inner) =>

{

Console.WriteLine("Exception is {0}", inner.Message);

return true;

});

}

Console.WriteLine("Press enter to finish");

Console.ReadLine();

在 .NET Framework 4 版中，当等待时间预计非常短时，并且当事件不会跨越进程边界时，可使用 System.Threading.ManualResetEventSlim 类以获得更好的性能。当等待事件变为已发出信号状态的过程中，ManualResetEventSlim 短时间内会使用繁忙旋转。 当等待时间很短时，旋转的开销相对于使用等待句柄来进行等待的开销会少很多。但是，如果事件在某个时间段内没有变为已发出信号状态，则 ManualResetEventSlim 会采用常规的事件处理等待。

4. AutoResetEvent

它和ManualResetEventSlim类似，AutoResetEvent 类表示一个本地等待处理事件，在释放了单个等待线程以后，该事件会在终止时自动重置。代码如下：

var arEvent = new AutoResetEvent(false);

CancellationTokenSource tokenSource = new CancellationTokenSource();

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

{

Task.Factory.StartNew(() =>

{

while (!tokenSource.Token.IsCancellationRequested)

{

arEvent.WaitOne();

Console.WriteLine("Task {0} released", Task.CurrentId);

}

}, tokenSource.Token);

}

Task signallingTask = Task.Factory.StartNew(() =>

{

while (!tokenSource.Token.IsCancellationRequested)

{

tokenSource.Token.WaitHandle.WaitOne(500);

arEvent.Set();

Console.WriteLine("Event set");

}

}, tokenSource.Token);

Console.ReadLine();

tokenSource.Cancel();

Console.WriteLine("Press enter to finish");

Console.ReadLine();

5. SemaphoreSlim

对可同时访问资源或资源池的线程数加以限制的 Semaphore 的轻量级实现。其原理就是在初始化时指定同时访问资源的数量，当某个任务执行完毕之后，调用Release()方法释放资源，当某个任务需要执行时，则它可以调用Wait()方法，阻塞当前线程直到它获取到资源为止。需要注意的时，这里的资源不是指什么数据库连接，内存什么的，这里的资源可以理解成只是一个协调任务同步的标记或者信号而已。代码如下：

var semaphore = new SemaphoreSlim(2);

CancellationTokenSource tokenSource = new CancellationTokenSource();

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

{

Task.Factory.StartNew(() =>

{

while (true)

{

semaphore.Wait(tokenSource.Token);

Console.WriteLine("Task {0} released", Task.CurrentId);

}

}, tokenSource.Token);

}

Task signallingTask = Task.Factory.StartNew(() =>

{

while (!tokenSource.Token.IsCancellationRequested)

{

tokenSource.Token.WaitHandle.WaitOne(500);

semaphore.Release(2);

Console.WriteLine("Semaphore released");

}

tokenSource.Token.ThrowIfCancellationRequested();

}, tokenSource.Token);

Console.ReadLine();

tokenSource.Cancel();

Console.WriteLine("Press enter to finish");

Console.ReadLine();

可以看到，.Net 4中，提供了更多的同步方法。极大方便了，我们对任务或线程同步的处理。我们可以根据需要，选择自己合适的、熟悉的方法进行对同步处理。