并行编程 - 《C#编程指南》 - 严富坤的知识库专栏(yanfukun.com)

并行编程的基本概念
- 并行编程的意义
- 并行编程的类型
任务并行库（Task Parallel Library, TPL）
并行循环
- Parallel.For和Parallel.ForEach
- 并行循环的控制和异常处理
并行LINQ（PLINQ）
异步编程模型（async/await）
并发集合
- 并发集合的基本概念
- 并发集合的常用操作
线程同步
小结

并行编程是一种编程范式，通过并发执行多个任务来提高程序的性能和效率。随着多核处理器的普及，并行编程在现代软件开发中的重要性日益增加。C#提供了丰富的并行编程工具和库，包括任务并行库（Task Parallel Library, TPL）、并行LINQ（PLINQ）和异步编程模型（async/await）。本文将深入探讨C#中的并行编程，从基本概念到高级用法，全面解析并行编程的原理和机制，并结合实际案例，帮助读者掌握并行编程的精髓。

并行编程的基本概念

并行编程的意义

并行编程的主要目标是提高程序的性能和响应速度。通过将计算任务分解为多个子任务并并发执行，可以更充分地利用多核处理器的计算能力，加快任务的完成时间。并行编程在数据密集型和计算密集型应用中尤为重要，如科学计算、图像处理和大数据分析等。

并行编程的类型

并行编程可以分为以下几种类型：

数据并行（Data Parallelism）：将相同的操作应用于多个数据元素。
任务并行（Task Parallelism）：将不同的任务并发执行。
管道并行（Pipeline Parallelism）：将任务划分为多个阶段，并在不同阶段并行处理数据。

任务并行库（Task Parallel Library, TPL）

任务并行库的基本概念

任务并行库（TPL）是C#中用于简化并行编程的一组库。TPL基于任务（Task）的概念，通过Task类和Task并行操作（如Task.Run、Task.WhenAll等）提供了高效的并行编程支持。

using System;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        Task task1 = Task.Run(() => DoWork(1));
        Task task2 = Task.Run(() => DoWork(2));
        Task.WaitAll(task1, task2);
        Console.WriteLine("所有任务完成");
    }
    public static void DoWork(int taskId)
    {
        Console.WriteLine($"任务{taskId}开始");
        Task.Delay(1000).Wait();
        Console.WriteLine($"任务{taskId}完成");
    }
}

在这个例子中，我们使用Task.Run方法并发执行了两个任务，并使用Task.WaitAll方法等待所有任务完成。

创建和启动任务

创建和启动任务的常用方法包括Task.Run、Task.Factory.StartNew和new Task().Start。

using System;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        Task task1 = Task.Run(() => DoWork(1));
        Task task2 = Task.Factory.StartNew(() => DoWork(2));
        Task task3 = new Task(() => DoWork(3));
        task3.Start();
        Task.WaitAll(task1, task2, task3);
        Console.WriteLine("所有任务完成");
    }
    public static void DoWork(int taskId)
    {
        Console.WriteLine($"任务{taskId}开始");
        Task.Delay(1000).Wait();
        Console.WriteLine($"任务{taskId}完成");
    }
}

在这个例子中，我们展示了创建和启动任务的三种常用方法。

任务的状态和结果

通过Task类的属性和方法，可以获取任务的状态和结果。常用属性包括Status、IsCompleted、IsFaulted、IsCanceled等。获取任务结果可以使用Task<TResult>类的Result属性或await关键字。

using System;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        Task<int> task = Task.Run(() => Compute(10));
        Console.WriteLine($"任务状态：{task.Status}");
        int result = task.Result;
        Console.WriteLine($"任务结果：{result}");
        Console.WriteLine($"任务状态：{task.Status}");
    }
    public static int Compute(int value)
    {
        Task.Delay(1000).Wait();
        return value * value;
    }
}

在这个例子中，我们使用Task<TResult>类获取任务的计算结果。

任务的取消和异常处理

任务的取消和异常处理是并行编程中的重要问题。通过CancellationToken类，可以实现任务的取消。通过try-catch语句，可以处理任务中的异常。

using System;
using System.Threading;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        Task task = Task.Run(() => DoWork(cts.Token), cts.Token);
        Task.Delay(500).Wait();
        cts.Cancel();
        try
        {
            task.Wait();
        }
        catch (AggregateException ex)
        {
            foreach (var innerEx in ex.InnerExceptions)
            {
                Console.WriteLine($"异常：{innerEx.Message}");
            }
        }
        Console.WriteLine($"任务状态：{task.Status}");
    }
    public static void DoWork(CancellationToken token)
    {
        for (int i = 0; i < 10; i++)
        {
            token.ThrowIfCancellationRequested();
            Console.WriteLine($"工作{i}");
            Task.Delay(200).Wait();
        }
    }
}

在这个例子中，我们演示了任务的取消和异常处理。

并行循环

Parallel.For和Parallel.ForEach

Parallel.For和Parallel.ForEach是TPL提供的并行循环操作，用于并发执行循环体中的代码。

using System;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        Parallel.For(0, 10, i =>
        {
            Console.WriteLine($"工作{i}");
            Task.Delay(100).Wait();
        });
        Console.WriteLine("所有工作完成");
    }
}

在这个例子中，我们使用Parallel.For并发执行了一个循环体。

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        List<int> numbers = new List<int> { 1, 2, 3, 4, 5 };
        Parallel.ForEach(numbers, number =>
        {
            Console.WriteLine($"工作{number}");
            Task.Delay(100).Wait();
        });
        Console.WriteLine("所有工作完成");
    }
}

在这个例子中，我们使用Parallel.ForEach并发执行了一个列表中的每个元素。

并行循环的控制和异常处理

通过ParallelOptions类，可以控制并行循环的行为，如最大并行度、取消标记等。并行循环中的异常可以通过AggregateException类进行处理。

using System;
using System.Threading;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        ParallelOptions options = new ParallelOptions
        {
            MaxDegreeOfParallelism = 4,
            CancellationToken = cts.Token
        };
        Task.Run(() =>
        {
            Task.Delay(500).Wait();
            cts.Cancel();
        });
        try
        {
            Parallel.For(0, 10, options, i =>
            {
                options.CancellationToken.ThrowIfCancellationRequested();
                if (i == 5) throw new InvalidOperationException($"异常：工作{i}");
                Console.WriteLine($"工作{i}");
                Task.Delay(100).Wait();
            });
        }
        catch (AggregateException ex)
        {
            foreach (var innerEx in ex.InnerExceptions)
            {
                Console.WriteLine(innerEx.Message);
            }
        }
        Console.WriteLine("所有工作完成");
    }
}

在这个例子中，我们控制了并行循环的最大并行度，并处理了循环中的异常。

并行LINQ（PLINQ）

PLINQ的基本概念

并行LINQ（Parallel LINQ, PLINQ）是LINQ的一种并行化实现，用于在多核处理器上并发执行LINQ查询。PLINQ通过将查询操作并行化，提高了数据密集型任务的性能。

using System;
using System.Collections.Generic;
using System.Linq;
public class Program
{
    public static void Main(string[] args)
    {
        List<int> numbers = Enumerable.Range(1, 10).ToList();
        var query = numbers.AsParallel().Where(number => number % 2 == 0).Select(number => number * 2);
        foreach (var num in query)
        {
            Console.WriteLine(num);
        }
    }
}

在这个例子中，我们使用PLINQ并发执行了一个LINQ查询。

PLINQ查询的控制和优化

通过ParallelQuery类和ParallelExecutionMode

枚举，可以控制PLINQ查询的行为和优化查询性能。

using System;
using System.Collections.Generic;
using System.Linq;
public class Program
{
    public static void Main(string[] args)
    {
        List<int> numbers = Enumerable.Range(1, 100000).ToList();
        var query = numbers.AsParallel()
                           .WithDegreeOfParallelism(4)
                           .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                           .Where(number => number % 2 == 0)
                           .Select(number => number * 2);
        foreach (var num in query)
        {
            Console.WriteLine(num);
        }
    }
}

在这个例子中，我们通过设置并行度和执行模式控制了PLINQ查询的行为。

PLINQ查询中的异常处理

PLINQ查询中的异常可以通过AggregateException类进行处理。

using System;
using System.Collections.Generic;
using System.Linq;
public class Program
{
    public static void Main(string[] args)
    {
        List<int> numbers = Enumerable.Range(1, 10).ToList();
        try
        {
            var query = numbers.AsParallel().Select(number =>
            {
                if (number == 5) throw new InvalidOperationException($"异常：数字{number}");
                return number * 2;
            });
            foreach (var num in query)
            {
                Console.WriteLine(num);
            }
        }
        catch (AggregateException ex)
        {
            foreach (var innerEx in ex.InnerExceptions)
            {
                Console.WriteLine(innerEx.Message);
            }
        }
    }
}

在这个例子中，我们处理了PLINQ查询中的异常。

异步编程模型（async/await）

异步编程模型的基本概念

异步编程模型（async/await）是C#中用于简化异步编程的语言特性。通过async关键字和await关键字，可以将异步操作表示为同步代码，提高代码的可读性和维护性。

using System;
using System.Threading.Tasks;
public class Program
{
    public static async Task Main(string[] args)
    {
        await DoWorkAsync();
        Console.WriteLine("所有工作完成");
    }
    public static async Task DoWorkAsync()
    {
        Console.WriteLine("开始工作");
        await Task.Delay(1000);
        Console.WriteLine("工作完成");
    }
}

在这个例子中，我们使用async和await实现了一个简单的异步操作。

创建和运行异步方法

通过将方法声明为async，并返回Task或Task<TResult>，可以创建异步方法。通过await关键字，可以等待异步方法的完成。

using System;
using System.Threading.Tasks;
public class Program
{
    public static async Task Main(string[] args)
    {
        int result = await ComputeAsync(10);
        Console.WriteLine($"结果：{result}");
    }
    public static async Task<int> ComputeAsync(int value)
    {
        await Task.Delay(1000);
        return value * value;
    }
}

在这个例子中，我们创建了一个异步方法ComputeAsync，并在Main方法中等待其完成。

异步方法的取消和异常处理

通过CancellationToken类，可以实现异步方法的取消。通过try-catch语句，可以处理异步方法中的异常。

using System;
using System.Threading;
using System.Threading.Tasks;
public class Program
{
    public static async Task Main(string[] args)
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        Task task = DoWorkAsync(cts.Token);
        cts.CancelAfter(500);
        try
        {
            await task;
        }
        catch (OperationCanceledException ex)
        {
            Console.WriteLine($"任务取消：{ex.Message}");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"异常：{ex.Message}");
        }
    }
    public static async Task DoWorkAsync(CancellationToken token)
    {
        for (int i = 0; i < 10; i++)
        {
            token.ThrowIfCancellationRequested();
            Console.WriteLine($"工作{i}");
            await Task.Delay(200, token);
        }
    }
}

在这个例子中，我们实现了异步方法的取消和异常处理。

并发集合

并发集合的基本概念

并发集合是用于在多线程环境下安全地操作集合数据的集合类。常用的并发集合类包括ConcurrentDictionary、ConcurrentBag、ConcurrentQueue和ConcurrentStack等。

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        ConcurrentBag<int> bag = new ConcurrentBag<int>();
        Parallel.For(0, 10, i =>
        {
            bag.Add(i);
        });
        foreach (var item in bag)
        {
            Console.WriteLine(item);
        }
    }
}

在这个例子中，我们使用ConcurrentBag并发添加和访问数据。

并发集合的常用操作

并发集合提供了丰富的方法，用于在多线程环境下安全地操作集合数据。常用的方法包括Add、TryTake、TryPeek等。

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;
public class Program
{
    public static void Main(string[] args)
    {
        ConcurrentQueue<int> queue = new ConcurrentQueue<int>();
        Parallel.For(0, 10, i =>
        {
            queue.Enqueue(i);
        });
        int item;
        while (queue.TryDequeue(out item))
        {
            Console.WriteLine(item);
        }
    }
}

在这个例子中，我们使用ConcurrentQueue并发添加和移除数据。

线程同步

线程同步的基本概念

线程同步是指在多线程环境下协调线程的执行顺序，确保共享数据的一致性和正确性。常用的线程同步机制包括锁（lock）、信号量（Semaphore）、事件（EventWaitHandle）等。

使用锁进行线程同步

通过lock语句，可以确保某段代码在同一时刻只能由一个线程执行。

using System;
using System.Threading;
using System.Threading.Tasks;
public class Program
{
    private static readonly object _lock = new object();
    private static int _counter = 0;
    public static void Main(string[] args)
    {
        Parallel.For(0, 1000, i =>
        {
            IncrementCounter();
        });
        Console.WriteLine($"最终计数：{_counter}");
    }
    private static void IncrementCounter()
    {
        lock (_lock)
        {
            _counter++;
        }
    }
}

在这个例子中，我们使用lock语句保护了对计数器的访问，确保了计数器的正确性。

使用信号量进行线程同步

通过Semaphore类，可以控制对资源的访问，并允许多个线程同时访问。

using System;
using System.Threading;
using System.Threading.Tasks;
public class Program
{
    private static readonly SemaphoreSlim _semaphore = new SemaphoreSlim(3);
    public static void Main(string[] args)
    {
        Parallel.For(0, 10, i =>
        {
            AccessResource(i);
        });
        Console.WriteLine("所有任务完成");
    }
    private static void AccessResource(int taskId)
    {
        _semaphore.Wait();
        try
        {
            Console.WriteLine($"任务{taskId}开始访问资源");
            Task.Delay(1000).Wait();
            Console.WriteLine($"任务{taskId}结束访问资源");
        }
        finally
        {
            _semaphore.Release();
        }
    }
}

在这个例子中，我们使用SemaphoreSlim控制了对资源的访问，允许最多三个线程同时访问资源。

使用事件进行线程同步

通过EventWaitHandle类，可以实现线程之间的事件通知和同步。

using System;
using System.Threading;
using System.Threading.Tasks;
public class Program
{
    private static readonly EventWaitHandle _event = new AutoResetEvent(false);
    public static void Main(string[] args)
    {
        Task.Run(() => WaitForEvent());
        Console.WriteLine("按任意键触发事件");
        Console.ReadKey();
        _event.Set();
    }
    private static void WaitForEvent()
    {
        Console.WriteLine("等待事件");
        _event.WaitOne();
        Console.WriteLine("事件已触发");
    }
}

在这个例子中，我们使用AutoResetEvent实现了线程之间的事件通知。

小结

并行编程是C#中用于提高程序性能和效率的重要技术。通过任务并行库（TPL）、并行LINQ（PLINQ）、异步编程模型（async/await）、并发集合和线程同步机制，开发者可以高效地进行并行编程，充分利用多核处理器的计算能力。本文深入探讨了C#中的并行编程，从基本概念到高级用法，全面解析了并行编程的原理和机制，并结合实际案例展示了并行编程在各种场景中的应用。

掌握并行编程不仅能够提高代码的执行效率，还能够在复杂应用程序中发挥重要作用。希望本文能帮助读者更好地理解和掌握C#中的并行编程，在实际开发中充分利用这一强大的编程工具。通过对并行编程技术的深入理解和合理应用，可以编写出更加高效、可靠和稳定的程序。