.NET Core 中的并发编程

时间:2021-04-03 人气:0

并发编程 - 异步 vs. 多线程代码

并行编程是一个广泛的术语，我们应该通过观察异步方法和实际的多线程之间的差异展开探讨。尽管 .NET Core 使用了任务来表达同样的概念，一个关键的差异是内部处理的不同。调用线程在做其他事情时，异步方法在后台运行。这意味着这些方法是 I/O 密集型的，即他们大部分时间用于输入和输出操作，例如文件或网络访问。只要有可能，使用异步 I/O 方法代替同步操作很有意义。相同的时间，调用线程可以在处理桌面应用程序中的用户交互或处理服务器应用程序中的同时处理其他请求，而不仅仅是等待操作完成。

计算密集型的方法要求 CPU 周期工作，并且只能运行在他们专用的后台线程中。CPU 的核心数限制了并行运行时的可用线程数量。操作系统负责在剩余的线程之间切换，使他们有机会执行代码。这些方法仍然被并发地执行，却不必被并行地执行。尽管这意味着方法不是同时执行，却可以在其他方法暂停的时候执行。

并行 vs 并发

本文将在最后一段中重点介绍在 .NET Core中多线程并发编程。

任务并行库

.NET Framework 4 引入了任务并行库 (TPL) 作为编写并发代码的首选 API。.NET Core采用相同的编程模式。要在后台运行一段代码，需要将其包装成一个任务：

var backgroundTask = Task.Run(() => DoComplexCalculation(42));
// do other work
var result = backgroundTask.Result;

当需要返回结果时，Task.Run 方法接收一个函数 (Func) ；当不需要返回结果时，方法 Task.Run 接收一个动作 (Action) 。当然，所有的情况下都可以使用 lambda 表达式，就像我上面例子中调用带一个参数的长时间方法。线程池中的某个线程将会处理任务。.NET Core 的运行时包含一个默认调度程序，使用线程池来处理队列并执行任务。您可以通过派生 TaskScheduler 类实现自己的调度算法，代替默认的，但这超过本文的讨论范围。正如我们之前所见，我使用 Result 属性来合并被调用的后台线程。对于不需要返回结果的线程，我可以调用 Wait() 来代替。这两种方式都将被堵塞到后台任务完成。为了避免堵塞调用线程 ( 如在ASP.NET Core应用程序中) ，可以使用 await 关键字：

var backgroundTask = Task.Run(() => DoComplexCalculation(42));
// do other work
var result = await backgroundTask;

这样被调用的线程将被释放以便处理其他传入请求。一旦任务完成，一个可用的工作线程将会继续处理请求。当然，控制器动作方法必须是异步的:

public async Task<iactionresult> Index() { // method body }

处理异常

将两个线程合并在一起的时候，任务抛出的任何异常将被传递到调用线程中：

如果使用 Result 或 Wait() ，它们将被打包到 AggregateException 中。实际的异常将被抛出并存储在其 InnerException 属性中。

如果您使用 await，原来的异常将不会被打包。

在这两种情况下，调用堆栈的信息将保持不变。

取消任务

由于任务是可以长时间运行的，所以你可能想要有一个可以提前取消任务的选项。实现这个选项，需要在任务创建的时候传入取消的令牌 (token)，之后再使用令牌触发取消任务：

var tokenSource = new CancellationTokenSource();
var cancellableTask = Task.Run(() =>
{
for (int i = 0; i < 100; i++)
{
if (tokenSource.Token.IsCancellationRequested)
{
// clean up before exiting
tokenSource.Token.ThrowIfCancellationRequested();
}
// do long-running processing
}
return 42;
}, tokenSource.Token);
// cancel the task
tokenSource.Cancel();
try
{
await cancellableTask;
}
catch (OperationCanceledException e)
{
// handle the exception
}

实际上，为了提前取消任务，你需要检查任务中的取消令牌，并在需要取消的时候作出反应:在执行必要的清理操作后，调用 ThrowIfCancellationRequested() 退出任务。这个方法将会抛出 OperationCanceledException，以便在调用线程中执行相应的处理。

协调多任务

如果你需要运行多个后台任务，这里有些方法可以帮助到你。要同时运行多个任务，只需连续启动它们并收集它们的引用，例如在数组中：

var backgroundTasks = new []
{
Task.Run(() => DoComplexCalculation(1)),
Task.Run(() => DoComplexCalculation(2)),
Task.Run(() => DoComplexCalculation(3))
};

现在你可以使用 Task 类的静态方法，等待他们被异步或者同步执行完毕。

// wait synchronously
Task.WaitAny(backgroundTasks);
Task.WaitAll(backgroundTasks);
// wait asynchronously
await Task.WhenAny(backgroundTasks);
await Task.WhenAll(backgroundTasks);

实际上，这两个方法最终都会返回所有任务的自身，可以像任何其他任务一样再次操作。为了获取对应任务的结果，你可以检查该任务的 Result 属性。处理多任务的异常有点棘手。方法 WaitAll 和 WhenAll 不管哪个任务被收集到异常时都会抛出异常。不过，对于 WaitAll ，将会收集所有的异常到对应的 InnerExceptions 属性；对于 WhenAll ，只会抛出第一个异常。为了确认哪个任务抛出了哪个异常，您需要单独检查每个任务的 Status 和 Exception 属性。在使用 WaitAny 和 WhenAny 时必须足够小心。他们会等到第一个任务完成 (成功或失败)，即使某个任务出现异常时也不会抛出任何异常。他们只会返回已完成任务的索引或者分别返回已完成的任务。你必须等到任务完成或访问其 result 属性时捕获异常，例如：

var completedTask = await Task.WhenAny(backgroundTasks);
try
{
var result = await completedTask;
}
catch (Exception e)
{
// handle exception
}

如果你想连续运行多个任务，代替并发任务，可以使用延续 (continuations)的方式：

var compositeTask = Task.Run(() => DoComplexCalculation(42))
.ContinueWith(previous => DoAnotherComplexCalculation(previous.Result),
TaskContinuationOptions.OnlyOnRanToCompletion)

ContinueWith() 方法允许你把多个任务一个接着一个执行。这个延续的任务将获取到前面任务的结果或状态的引用。你仍然可以增加条件判断是否执行延续任务，例如只有在前面任务成功执行或者抛出异常时。对比连续等待多个任务，提高了灵活性。当然，您可以将延续任务与之前讨论的所有功能相结合：异常处理、取消和并行运行任务。这就有了很大的表演空间，以不同的方式进行组合：

var multipleTasks = new[]
{
Task.Run(() => DoComplexCalculation(1)),
Task.Run(() => DoComplexCalculation(2)),
Task.Run(() => DoComplexCalculation(3))
};
var combinedTask = Task.WhenAll(multipleTasks);
var successfulContinuation = combinedTask.ContinueWith(task =>
CombineResults(task.Result), TaskContinuationOptions.OnlyOnRanToCompletion);
var failedContinuation = combinedTask.ContinueWith(task =>
HandleError(task.Exception), TaskContinuationOptions.NotOnRanToCompletion);
await Task.WhenAny(successfulContinuation, failedContinuation);

任务同步

如果任务是完全独立的，那么我们刚才看到的协调方法就已足够。然而，一旦需要同时共享数据，为了防止数据损坏，就必须要有额外的同步。两个以及更多的线程同时更新一个数据结构时，数据很快就会变得不一致。就好像下面这个示例代码一样：

var counters = new Dictionary< int, int >();
if (counters.ContainsKey(key))
{
counters[key] ++;
}
else
{
counters[key] = 1;
}

当多个线程同时执行上述代码时，不同线程中的特定顺序执行指令可能导致数据不正确，例如：

所有线程将会检查集合中是否存在同一个 key
结果，他们都会进入 else 分支，并将这个 key 的值设为1
最后结果将会是1，而不是2。如果是接连着执行代码的话，将会是预期的结果。

上述代码中，临界区 (critical section) 一次只允许一个线程可以进入。在C# 中，可以使用 lock 语句来实现:

var counters = new Dictionary< int, int >();
lock (syncObject)
{
if (counters.ContainsKey(key))
{
counters[key]++;
}
else
{
counters[key] = 1;
}
}

在这个方法中，所有线程都必须共享相同的的 syncObject 。作为最佳做法，syncObject 应该是一个专用的 Object 实例，专门用于保护对一个独立的临界区的访问，避免从外部访问。在 lock 语句中，只允许一个线程访问里面的代码块。它将阻止下一个尝试访问它的线程，直到前一个线程退出。这将确保线程完整执行临界区代码，而不会被另一个线程中断。当然，这将减少并行性并减慢代码的整体执行速度，因此您最好最小化临界区的数量并使其尽可能的短。

使用 Monitor 类来简化 lock 声明：

var lockWasTaken = false;
var temp = syncObject;
try
{
Monitor.Enter(temp, ref lockWasTaken);
// lock statement body
}
finally
{
if (lockWasTaken)
{
Monitor.Exit(temp);
}
}

尽管大部分时间您都希望使用 lock 语句，但 Monitor 类可以在需要时给予额外的控制。例如，您可以使用 TryEnter() 而不是 Enter()，并指定一个限定时间，避免无止境地等待锁释放。

其他同步基元

Monitor 只是 .NET Core 中众多同步基元的一员。根据实际情况，其他基元可能更适合。

Mutex 是 Monitor 更重量级的版本，依赖于底层的操作系统，提供跨多个进程同步访问资源[1]，是针对 Mutex 进行同步的推荐替代方案。

SemaphoreSlim 和 Semaphore 可以限制同时访问资源的最大线程数量，而不是像 Monitor 一样只能限制一个线程。 SemaphoreSlim 比 Semaphore 更轻量，但仅限于单个进程。如果可能，您最好使用 SemaphoreSlim 而不是 Semaphore。

ReaderWriterLockSlim 可以区分两种对访问资源的方式。它允许无限数量的读取器 (readers) 同时访问资源，并且限制同时只允许一个写入器 (writers) 访问锁定资源。读取时线程安全，但修改数据时需要独占资源，很好地保护了资源。

AutoResetEvent、ManualResetEvent 和 ManualResetEventSlim 将堵塞传入的线程，直到它们接收到一个信号 (即调用 Set() )。然后等待中的线程将继续执行。AutoResetEvent 在下一次调用 Set() 之前，将一直阻塞，并只允许一个线程继续执行。ManualResetEvent 和 ManualResetEventSlim 不会堵塞线程，除非 Reset() 被调用。ManualResetEventSlim 比前两者更轻量，更值得推荐。

Interlocked 提供一种选择——原子操作，这是替代 locking 和其他同步基元更好的选择（如果适用）：

// non-atomic operation with a lock
lock (syncObject)
{
counter++;
}
// equivalent atomic operation that doesn't require a lock
Interlocked.Increment(ref counter);

并发集合

当一个临界区需要确保对数据结构的原子访问时，用于并发访问的专用数据结构可能是更好和更有效的替代方案。例如，使用 ConcurrentDictionary 而不是 Dictionary，可以简化 lock 语句示例：

var counters = new ConcurrentDictionary< int, int >();
counters.TryAdd(key, 0);
lock (syncObject)
{
counters[key]++;
}

自然地，也有可能像下面一样:

counters.AddOrUpdate(key, 1, (oldKey, oldValue) => oldValue + 1);

因为 update 的委托是临界区外面的方法，因此，第二个线程可能在第一个线程更新值之前，读取到同样的旧值，使用自己的值有效地覆盖了第一个线程的更新值，这就丢失了一个增量。错误使用并发集合也是无法避免多线程带来的问题。并发集合的另一个替代方案是不变的集合 (immutable collections)。类似于并发集合，同样是线程安全的，但是底层实现是不一样的。任何关改变数据结构的操作将不会改变原来的实例。相反，它们返回一个更改后的副本，并保持原始实例不变：

var original = new Dictionary< int, int >().ToImmutableDictionary();
var modified = original.Add(key, value);

因此在一个线程中对集合任何更改对于其他线程来说都是不可见的。因为它们仍然引用原来的未修改的集合，这就是不变的集合本质上是线程安全的原因。当然，这使得它们对于解决不同集合的问题很有效。最好的情况是多个线程在同一个输入集合的情况下，独立地修改数据，在最后一步可能为所有线程合并变更。而使用常规集合，需要提前为每个线程创建集合的副本。

并行LINQ (PLINQ)

并行LINQ (PLINQ) 是 Task Parallel Library 的替代方案。顾名思义，它很大程度上依赖于 LINQ（语言集成查询）功能。对于在大集合中执行相同的昂贵操作的场景是很有用的。与所有操作都是顺序执行的普通 LINQ to Objects 不同的是，PLINQ可以在多个CPU上并行执行这些操作。发挥优势所需要的代码改动也是极小的:

// sequential execution
var sequential = Enumerable.Range(0, 40)
.Select(n => ExpensiveOperation(n))
.ToArray();
// parallel execution
var parallel = Enumerable.Range(0, 40)
.AsParallel()
.Select(n => ExpensiveOperation(n))
.ToArray();

如你所见，这两个代码片段的不同仅仅是调用 AsParallel()。这将IEnumerable 转换为 ParallelQuery，导致查询的部分并行运行。要切换为回顺序执行，您可以调用 AsSequential()，它将再次返回一个IEnumerable。默认情况下，PLINQ 不保留集合中的顺序，以便让进程更有效率。但是当顺序很重要时，可以调用 AsOrdered():

var parallel = Enumerable.Range(0, 40)
.AsParallel()
.AsOrdered()
.Select(n => ExpensiveOperation(n))
.ToArray();

同理，你可以通过调用 AsUnordered() 切换回来。

在完整的 .NET Framework 中并发编程

由于 .NET Core 是完整的 .NET Framework 的简化实现，所以 .NET Framework 中所有并行编程方法也可以在.NET Core 中使用。唯一的例外是不变的集合，它们不是完整的 .NET Framework 的组成部分。它们作为单独的 NuGet 软件包(System.Collections.Immutable)分发，您需要在项目中安装使用。

结论：

每当应用程序包含可以并行运行的 CPU 密集型代码时，利用并发编程来提高性能并提高硬件利用率是很有意义的。 .NET Core 中的 API 抽象了许多细节，使编写并发代码更容易。然而需要注意某些潜在的问题，其中大部分涉及从多个线程访问共享数据。如果可以的话，你应该完全避免这种情况。如果不行，请确保选择最合适的同步方法或数据结构。

您可能感兴趣的文章:

加载全部内容