C#多线程(9):多阶段并行线程

内容纲要

前言

这一篇,我们将学习用于实现并行任务、使得多个线程有序同步完成多个阶段的任务。

应用场景主要是控制 N 个线程(可随时增加或减少执行的线程),使得多线程在能够在 M 个阶段中保持同步。

线程工作情况如下:

《C#多线程(9):多阶段并行线程》

我们接下来 将学习C# 中的 Barrier ,用于实现并行协同工作。

Barrier 类

使多个任务能够采用并行方式依据某种算法在多个阶段中协同工作,使多个线程(称为“参与者” )分阶段同时处理算法

可以使多个线程(称为“参与者” )分阶段同时处理算法。(注意算法这个词)

每个参与者完成阶段任务后后将被阻止继续执行,直至所有参与者都已达到同一阶段。

Barrier 的构造函数如下:

构造函数 说明
Barrier(Int32) 初始化 Barrier 类的新实例。
Barrier(Int32, Action) 初始化 Barrier 类的新实例。

其中一个构造函数定义如下:

public Barrier (int participantCount, Action postPhaseAction);

participantCount :处于的线程数量,大于0并且小于32767。

postPhaseAction :在每个阶段后执行 Action(委托)。

属性和方法

在还没有清楚这个类有什么作用前,我们来看一下这个类的常用属性和方法。

大概了解 Barrier 有哪些常用属性和方法后,我们开始编写示例代码。

属性:

属性 说明
CurrentPhaseNumber 获取屏障的当前阶段的编号。
ParticipantCount 获取屏障中参与者的总数。
ParticipantsRemaining 获取屏障中尚未在当前阶段发出信号的参与者的数量。

方法:

方法 说明
AddParticipant() 通知 Barrier,告知其将会有另一个参与者。
AddParticipants(Int32) 通知 Barrier,告知其将会有多个其他参与者。
RemoveParticipant() 通知 Barrier,告知其将会减少一个参与者。
RemoveParticipants(Int32) 通知 Barrier,告知其将会减少一些参与者。
SignalAndWait() 发出参与者已达到屏障并等待所有其他参与者也达到屏障。
SignalAndWait(CancellationToken) 发出参与者已达到屏障的信号,并等待所有其他参与者达到屏障,同时观察取消标记。
SignalAndWait(Int32) 发出参与者已达到屏障的信号,并等待所有其他参与者也达到屏障,同时使用 32 位带符号整数测量超时。
SignalAndWait(Int32, CancellationToken) 发出参与者已达到屏障的信号,并等待所有其他参与者也达到屏障,使用 32 位带符号整数测量超时,同时观察取消标记。
SignalAndWait(TimeSpan) 发出参与者已达到屏障的信号,并等待所有其他参与者也达到屏障,同时使用 TimeSpan 对象测量时间间隔。
SignalAndWait(TimeSpan, CancellationToken) 发出参与者已达到屏障的信号,并等待所有其他参与者也达到屏障,使用 TimeSpan 对象测量时间间隔,同时观察取消标记。

Barrier 翻译屏障,前面所说的 “阶段”,在文档中称为屏障,官方有一些例子和实践场景:

https://docs.microsoft.com/zh-cn/dotnet/standard/threading/barrier?view=netcore-3.1

https://docs.microsoft.com/zh-cn/dotnet/standard/threading/how-to-synchronize-concurrent-operations-with-a-barrier?view=netcore-3.1

本文的教程比较简单,你可以先看本教程,再去看看官方示例。

示例

假设有个比赛,一个有三个环节,有三个小组参加比赛。

比赛有三个环节,小组完成一个环节后,可以去等待区休息,等待其他小组也完成比赛后,开始进行下一个环节的比赛。

示例如下:

new Barrier(int,Action) 设置有多少线程参与,Action 委托设置每个阶段完成后执行哪些动作。

.SignalAndWait() 阻止当前线程继续往下执行;直到其他完成也执行到此为止。

    class Program
    {
        // Barrier(Int32, Action)
        private static Barrier barrier = new Barrier(3, b =>
                            Console.WriteLine($"\n第 {b.CurrentPhaseNumber + 1} 环节的比赛结束,请评分!"));

        static void Main(string[] args)
        {
            // Random 模拟每个小组完成一个环节比赛需要的时间
            Thread thread1 = new Thread(() => DoWork("第一小组", new Random().Next(2, 10)));
            Thread thread2 = new Thread(() => DoWork("第二小组", new Random().Next(2, 10)));
            Thread thread3 = new Thread(() => DoWork("第三小组", new Random().Next(2, 10)));

            // 三个小组开始比赛
            thread1.Start();
            thread2.Start();
            thread3.Start();

            Console.ReadKey();
        }
        static void DoWork(string name, int seconds)
        {
            // 第一环节
            Console.WriteLine($"\n{name}:开始进入第一环节比赛");
            Thread.Sleep(TimeSpan.FromSeconds(seconds));    // 模拟小组完成环节比赛需要的时间
            Console.WriteLine($"\n    {name}:完成第一环节比赛,等待其它小组");
            // 小组完成阶段任务,去休息等待其它小组也完成比赛
            barrier.SignalAndWait();

            // 第二环节
            Console.WriteLine($"\n        {name}:开始进入第二环节比赛");
            Thread.Sleep(TimeSpan.FromSeconds(seconds));
            Console.WriteLine($"\n        {name}:完成第二环节比赛,等待其它小组\n");
            barrier.SignalAndWait();

            // 第三环节
            Console.WriteLine($"\n        {name}:开始进入第三环节比赛");
            Thread.Sleep(TimeSpan.FromSeconds(seconds));
            Console.WriteLine($"\n        {name}:完成第三环节比赛,等待其它小组\n");
            barrier.SignalAndWait();
        }
    }

上面的示例中,每个线程都使用了 DoWork() 这个方法去中相同的事情,当然也可以设置多个线程执行不同的任务,但是必须保证每个线程都具有相同数量的 .SignalAndWait(); 方法。

当然 SignalAndWait() 可以设置等待时间,如果其他线程迟迟没有到这一步,那就继续运行。可以避免死锁等问题。

到目前,只使用了 SignalAndWait() ,我们继续学习一下 Barrier 类的其他方法。

新的示例

Barrier.AddParticipant():添加参与者;

Barrier.RemoveParticipant():移除参与者;

这里继续使用第二节的示例。

因为这是比赛,老是等待其他小组,会使得比赛进行比较慢。

新的规则:不必等待最后一名,当环节只剩下最后一名时为完成时,其它小组可以立即进行下一个环节的比赛。

​ 当然,最后一名小组,有权利继续完成比赛。

修改第二小节的代码,在 Main 内第一行加上 barrier.RemoveParticipant();

        static void Main(string[] args)
        {
            barrier.RemoveParticipant();
            ... ...

试着再运行一下。

说明

SignalAndWait() 的 重载比较多,例如 SignalAndWait(CancellationToken),这里笔者先不讲解此方法如何使用。等到写到后面的异步(Task),读者学到相关的知识点,我们再过一次复习,这样由易到难,自然水到渠成。

Barrier 适合用于同时执行相同流程的工作,因为工作内容是相同的,便于协同。工作流有可能用得上吧。

但是 Barrier 更加适合用于算法领域,可以参考:https://devblogs.microsoft.com/pfxteam/parallel-merge-sort-using-barrier/

当然,后面学习异步和并行编程后,也会编写相应的算法示例。

点赞

发表评论

电子邮件地址不会被公开。 必填项已用*标注