C#并行Task同步

springsnow 2022-06-03 我要评论

在并行计算中，不可避免的会碰到多个任务共享变量，实例，集合。虽然task自带了两个方法：task.ContinueWith()和Task.Factory.ContinueWhenAll()来实现任务串行化，但是这些简单的方法远远不能满足我们实际的开发需要，从.net 4.0开始，类库给我们提供了很多的类来帮助我们简化并行计算中复杂的数据同步问题。

一、隔离执行：不共享数据，让每个task都有一份自己的数据拷贝。

对数据共享问题处理的方式是“分离执行”，我们通过把每个Task执行完成后的各自计算的值进行最后的汇总，也就是说多个Task之间不存在数据共享了，各自做各自的事，完全分离开来。

1、传统方式

每个Task执行时不存在数据共享了，每个Task中计算自己值，最后我们汇总每个Task的Result。我们可以通过Task中传递的state参数来进行隔离执行：

int Sum = 0;
Task<int>[] tasks = new Task<int>[10];
for (int i = 0; i < 10; i++)
{
    tasks[i] = new Task<int>((obj) =>
    {
        var start = (int)obj;
        for (int j = 0; j < 1000; j++)
        {
            start = start + 1;
        }
        return start;
    }, Sum);
    tasks[i].Start();
}
Task.WaitAll(tasks);
for (var i = 0; i < 10; i++)
{
    Sum += tasks[i].Result;
}
Console.WriteLine("Expected value {0}, Parallel value: {1}", 10000, Sum);

2、ThreadLocal类

在.Net中提供了System.Threading.ThreadLocal来创建分离。

ThreadLocal是一种提供线程本地存储的类型，它可以给每个线程一个分离的实例，来提供每个线程单独的数据结果。上面的程序我们可以使用TreadLocal：

int Sum = 0;
Task<int>[] tasks = new Task<int>[10];

var tl = new ThreadLocal<int>();
for (int i = 0; i < 10; i++)
{
    tasks[i] = new Task<int>((obj) =>
    {
        tl.Value = (int)obj;
        for (int j = 0; j < 1000; j++)
        {
            tl.Value++;
        }
        returntl.Value;
    }, Sum);
    tasks[i].Start();
}
Task.WaitAll(tasks);
for (var i = 0; i < 10; i++)
{
    Sum += tasks[i].Result;
}
Console.WriteLine("Expected value {0}, Parallel value: {1}", 10000, Sum);

但是我们要注意的一点TreadLocal是针对每个线程的，不是针对每个Task的。一个Tread中可能有多个Task。

ThreadLocal类举例：

static ThreadLocal<string> local;
static void Main()
{
    //创建ThreadLocal并提供默认值 
    local = new ThreadLocal<string>(() => "hehe");

    //修改TLS的线程 
    Thread th = new Thread(() =>
     {
         local.Value = "Mgen";
         Display();
     });

    th.Start();
    th.Join();
    Display();
}

//显示TLS中数据值 
static void Display()
{
    Console.WriteLine("{0} {1}", Thread.CurrentThread.ManagedThreadId, local.Value);
}

二、同步类型：通过调整task的执行，有序的执行task。

同步类型是一种用来调度Task访问临界区域的一种特殊类型。在.Net 4.0中提供了多种同步类型给我们使用，主要分为：轻量级的、重量级的和等待处理型的，在下面我们会介绍常用的同步处理类型。

常用的同步类型

首先来看看.Net 4.0中常见的几种同步类型以及处理的相关问题：

同步类型以及解决问题

lock关键字、Montor类、SpinLock类：有序访问临界区域
Interlocked类：数值类型的增加或则减少
Mutex类：交叉同步
WaitAll方法：同步多个锁定（主要是Task之间的调度）
申明性的同步（如Synchronization）：使类中的所有的方法同步

1、Lock锁

其实最简单同步类型的使用办法就是使用lock关键字。在使用lock关键字时，首先我们需要创建一个锁定的object，而且这个object需要所有的task都能访问，其次能我们需要将我们的临界区域包含在lock块中。我们之前例子中代码可以这样加上lock:

int Sum = 0;
Task[] tasks = new Task[10];

var obj = new Object(); for (int i = 0; i < 10; i++)
{
    tasks[i] = new Task(() =>
    {
        for (int j = 0; j < 1000; j++)
        {
            lock (obj)
            { Sum = Sum + 1; }
        }
    });
    tasks[i].Start();
}
Task.WaitAll(tasks);
Console.WriteLine("Expected value {0}, Parallel value: {1}", 10000, Sum);

其实lock关键字是使用Monitor的一种简短的方式，lock关键字自动通过调用Monitor.Enter\Monitor.Exit方法来处理获得锁以及释放锁。

2、Interlocked 联锁

Interlocked通过使用操作系统或则硬件的一些特性提供了一些列高效的静态的同步方法。其中主要提供了这些方法：Exchange、Add、Increment、CompareExchange四种类型的多个方法的重载。我们将上面的例子中使用Interlocked:

int Sum = 0;
Task[] tasks = new Task[10];
for (int i = 0; i < 10; i++)
{
    tasks[i] = new Task(() =>
    {
        for (int j = 0; j < 1000; j++)
        {
            Interlocked.Increment(ref Sum);
        }
    });
    tasks[i].Start();
}
Task.WaitAll(tasks);
Console.WriteLine("Expected value {0}, Parallel value: {1}", 10000, Sum);

3、Mutex互斥体

Mutex也是一个同步类型，在多个线程进行访问的时候，它只向一个线程授权共享数据的独立访问。我们可以通过Mutex中的WaitOne方法来获取Mutex的所有权，但是同时我们要注意的是，我们在一个线程中多少次调用过WaitOne方法，就需要调用多少次ReleaseMutex方法来释放Mutex的占有。上面的例子我们通过Mutex这样实现：

int Sum = 0;
Task[] tasks = new Task[10];

var mutex = new Mutex();
for (int i = 0; i < 10; i++)
{
    tasks[i] = new Task(() =>
    {
        for (int j = 0; j < 1000; j++)
        {
            bool lockAcquired = mutex.WaitOne(); try
            {
                Sum++;
            }
            finally
            {
                if (lockAcquired) mutex.ReleaseMutex();
            }
        }
    });
    tasks[i].Start();
}
Task.WaitAll(tasks);
Console.WriteLine("Expected value {0}, Parallel value: {1}", 10000, Sum);

三、申明性同步

我们可以通过使用Synchronization 特性来标识一个类，从而使一个类型的字段以及方法都实现同步化。在使用Synchronization 时，我们需要将我们的目标同步的类继承于System.ContextBoundObject类型。我们来看看之前的例子我们同步标识Synchronization 的实现：

static void Main(string[] args)
{
    var sum = new SumClass();
    Task[] tasks = new Task[10];
    for (int i = 0; i < 10; i++)
    {
        tasks[i] = new Task(() =>
        {
            for (int j = 0; j < 1000; j++)
            {
                sum.Increment();
            }
        });
        tasks[i].Start();
    }
    Task.WaitAll(tasks);
    Console.WriteLine("Expected value {0}, Parallel value: {1}", 10000, sum.GetSum());
}

[Synchronization]
class SumClass : ContextBoundObject
{
    private int _Sum;

    public void Increment()
    {
        _Sum++;
    }

    public int GetSum()
    {
        return _Sum;
    }
}

四、并发集合

当多个线程对某个非线程安全容器并发地进行读写操作时，这些操作将导致不可预估的后果或者会导致报错。为了解决这个问题我们可以使用lock关键字或者Monitor类来给容器上锁。但锁的引入使得我们的代码更加复杂，同时也带来了更多的同步消耗。而.NET Framework 4提供的线程安全且可拓展的并发集合能够使得我们的并行代码更加容易编写，此外，锁的使用次数的减少也减少了麻烦的死锁与竞争条件的问题。.NET Framework 4主要提供了如下几种并发集合：BlockingCollection，ConcurrentBag，ConcurrentDictionary，ConcurrentQueue，ConcurrentStack。这些集合通过使用一种叫做比较并交换(compare and swap, CAS)指令和内存屏障的技术来避免使用重量级的锁。

在.Net 4.0中提供了很多并发的集合类型来让我们处理数据同步的集合的问题，这里面包括：

1.ConcurrentQueue：提供并发安全的队列集合，以先进先出的方式进行操作；
2.ConcurrentStack：提供并发安全的堆栈集合，以先进后出的方式进行操作；
3.ConcurrentBag：提供并发安全的一种无序集合；
4.ConcurrentDictionary：提供并发安全的一种key-value类型的集合。

我们在这里只做ConcurrentQueue的一个尝试，并发队列是一种线程安全的队列集合，我们可以通过Enqueue()进行排队、TryDequeue()进行出队列操作：

for (var j = 0; j < 10; j++)
{
    var queue = new ConcurrentQueue<int>();
    var count = 0;
    for (var i = 0; i < 1000; i++)
    {
        queue.Enqueue(i);
    }
    var tasks = new Task[10];
    for (var i = 0; i < tasks.Length; i++)
    {
        tasks[i] = new Task(() =>
        {
            while (queue.Count > 0)
            {
                int item;
                var isDequeue = queue.TryDequeue(out item);
                if (isDequeue) Interlocked.Increment(ref count);
            }

        });
        tasks[i].Start();
    }
    try
    {
        Task.WaitAll(tasks);
    }
    catch (AggregateException e)
    {
        e.Handle((ex) =>
        {
            Console.WriteLine("Exception Message:{0}", ex.Message);
            return true;
        });
    }
    Console.WriteLine("Dequeue items count :{0}", count);
}

五、Barrier（屏障同步）

barrier叫做屏障，就像下图中的“红色线”，如果我们的屏障设为4个task就认为已经满了的话，那么执行中先到的task必须等待后到的task，通知方式也就是barrier.SignalAndWait()，屏障中线程设置操作为new Barrier(4,(i)=>{})。SignalAndWait给我们提供了超时的重载，为了能够取消后续执行

//四个task执行
static Task[] tasks = new Task[4];

static Barrier barrier = null;

static void Main(string[] args)
{
    barrier = new Barrier(tasks.Length, (i) =>
     {
         Console.WriteLine("**********************************************************");
         Console.WriteLine("\n屏障中当前阶段编号:{0}\n", i.CurrentPhaseNumber);
         Console.WriteLine("**********************************************************");
     });

    for (int j = 0; j < tasks.Length; j++)
    {
        tasks[j] = Task.Factory.StartNew((obj) =>
        {
            var single = Convert.ToInt32(obj);

            LoadUser(single);
            barrier.SignalAndWait();

            LoadProduct(single);
            barrier.SignalAndWait();

            LoadOrder(single);
            barrier.SignalAndWait();
        }, j);
    }

    Task.WaitAll(tasks);

    Console.WriteLine("指定数据库中所有数据已经加载完毕！");

    Console.Read();
}

static void LoadUser(int num)
{
    Console.WriteLine("当前任务:{0}正在加载User部分数据！", num);
}

static void LoadProduct(int num)
{
    Console.WriteLine("当前任务:{0}正在加载Product部分数据！", num);
}

static void LoadOrder(int num)
{
    Console.WriteLine("当前任务:{0}正在加载Order部分数据！", num);
}