〇、前言
日常开发中经常会遇到数据统计,特别是关于报表的项目。数据处理的效率和准确度当然是首要关注点。
本文主要介绍,如何通过 Parallel 来并行处理数据,并组合 ConcurrentBag 集合,来将处理效率达到高点的同时,也能确保数据的准确。
一、ConcurrentBag 简介
1、简介与源码
ConcurrentBag,表示对象的线程安全的无序集合。ConcurrentBag 内部将数据按线程的标识独立进行存储,程序可以在同一个线程中插入、删除元素,所以每个线程对其数据的操作是非常快的。
下面是源码供参考:
点击展开 ConcurrentBag 源码
// System.Collections.Concurrent, Version=5.0.0.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a
// System.Collections.Concurrent.ConcurrentBag
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Threading;
[DebuggerTypeProxy(typeof(System.Collections.Concurrent.IProducerConsumerCollectionDebugView))]
[DebuggerDisplay("Count = {Count}")]
public class ConcurrentBag : IProducerConsumerCollection, IEnumerable, IEnumerable, ICollection, IReadOnlyCollection
{
private sealed class WorkStealingQueue
{
private volatile int _headIndex;
private volatile int _tailIndex;
private volatile T[] _array = new T[32];
private volatile int _mask = 31;
private int _addTakeCount;
private int _stealCount;
internal volatile int _currentOp;
internal bool _frozen;
internal readonly WorkStealingQueue _nextQueue;
internal readonly int _ownerThreadId;
internal bool IsEmpty => _headIndex - _tailIndex >= 0;
internal int DangerousCount
{
get
{
int stealCount = _stealCount;
int addTakeCount = _addTakeCount;
return addTakeCount - stealCount;
}
}
internal WorkStealingQueue(WorkStealingQueue nextQueue)
{
_ownerThreadId = Environment.CurrentManagedThreadId;
_nextQueue = nextQueue;
}
internal void LocalPush(T item, ref long emptyToNonEmptyListTransitionCount)
{
bool lockTaken = false;
try
{
Interlocked.Exchange(ref _currentOp, 1);
int num = _tailIndex;
if (num == int.MaxValue)
{
_currentOp = 0;
lock (this)
{
_headIndex &= _mask;
num = (_tailIndex = num & _mask);
Interlocked.Exchange(ref _currentOp, 1);
}
}
int headIndex = _headIndex;
if (!_frozen && headIndex - (num - 1) = _mask)
{
T[] array = new T[_array.Length = 0)
{
result = default(T);
return false;
}
bool lockTaken = false;
try
{
_currentOp = 2;
Interlocked.Exchange(ref _tailIndex, --tailIndex);
if (!_frozen && _headIndex - tailIndex = 0 && _currentOp == 1)
{
SpinWait spinWait = default(SpinWait);
do
{
spinWait.SpinOnce();
}
while (_currentOp == 1);
}
Interlocked.Exchange(ref _headIndex, headIndex + 1);
if (headIndex = arrayIndex; num--)
{
array[num] = _array[headIndex++ & _mask];
}
return dangerousCount;
}
}
private sealed class Enumerator : IEnumerator, IDisposable, IEnumerator
{
private readonly T[] _array;
private T _current;
private int _index;
public T Current => _current;
object IEnumerator.Current
{
get
{
if (_index == 0 || _index == _array.Length + 1)
{
throw new InvalidOperationException(System.SR.ConcurrentBag_Enumerator_EnumerationNotStartedOrAlreadyFinished);
}
return Current;
}
}
public Enumerator(T[] array)
{
_array = array;
}
public bool MoveNext()
{
if (_index _locals;
private volatile WorkStealingQueue _workStealingQueues;
private long _emptyToNonEmptyListTransitionCount;
public int Count
{
get
{
if (_workStealingQueues == null)
{
return 0;
}
bool lockTaken = false;
try
{
FreezeBag(ref lockTaken);
return DangerousCount;
}
finally
{
UnfreezeBag(lockTaken);
}
}
}
private int DangerousCount
{
get
{
int num = 0;
for (WorkStealingQueue workStealingQueue = _workStealingQueues; workStealingQueue != null; workStealingQueue = workStealingQueue._nextQueue)
{
num = checked(num + workStealingQueue.DangerousCount);
}
return num;
}
}
public bool IsEmpty
{
get
{
WorkStealingQueue currentThreadWorkStealingQueue = GetCurrentThreadWorkStealingQueue(forceCreate: false);
if (currentThreadWorkStealingQueue != null)
{
if (!currentThreadWorkStealingQueue.IsEmpty)
{
return false;
}
if (currentThreadWorkStealingQueue._nextQueue == null && currentThreadWorkStealingQueue == _workStealingQueues)
{
return true;
}
}
bool lockTaken = false;
try
{
FreezeBag(ref lockTaken);
for (WorkStealingQueue workStealingQueue = _workStealingQueues; workStealingQueue != null; workStealingQueue = workStealingQueue._nextQueue)
{
if (!workStealingQueue.IsEmpty)
{
return false;
}
}
}
finally
{
UnfreezeBag(lockTaken);
}
return true;
}
}
bool ICollection.IsSynchronized => false;
object ICollection.SyncRoot
{
get
{
throw new NotSupportedException(System.SR.ConcurrentCollection_SyncRoot_NotSupported);
}
}
private object GlobalQueuesLock => _locals;
public ConcurrentBag()
{
_locals = new ThreadLocal();
}
public ConcurrentBag(IEnumerable collection)
{
if (collection == null)
{
throw new ArgumentNullException("collection", System.SR.ConcurrentBag_Ctor_ArgumentNullException);
}
_locals = new ThreadLocal();
WorkStealingQueue currentThreadWorkStealingQueue = GetCurrentThreadWorkStealingQueue(forceCreate: true);
foreach (T item in collection)
{
currentThreadWorkStealingQueue.LocalPush(item, ref _emptyToNonEmptyListTransitionCount);
}
}
public void Add(T item)
{
GetCurrentThreadWorkStealingQueue(forceCreate: true).LocalPush(item, ref _emptyToNonEmptyListTransitionCount);
}
bool IProducerConsumerCollection.TryAdd(T item)
{
Add(item);
return true;
}
public bool TryTake([MaybeNullWhen(false)] out T result)
{
WorkStealingQueue currentThreadWorkStealingQueue = GetCurrentThreadWorkStealingQueue(forceCreate: false);
if (currentThreadWorkStealingQueue == null || !currentThreadWorkStealingQueue.TryLocalPop(out result))
{
return TrySteal(out result, take: true);
}
return true;
}
public bool TryPeek([MaybeNullWhen(false)] out T result)
{
WorkStealingQueue currentThreadWorkStealingQueue = GetCurrentThreadWorkStealingQueue(forceCreate: false);
if (currentThreadWorkStealingQueue == null || !currentThreadWorkStealingQueue.TryLocalPeek(out result))
{
return TrySteal(out result, take: false);
}
return true;
}
private WorkStealingQueue GetCurrentThreadWorkStealingQueue(bool forceCreate)
{
WorkStealingQueue workStealingQueue = _locals.Value;
if (workStealingQueue == null)
{
if (!forceCreate)
{
return null;
}
workStealingQueue = CreateWorkStealingQueueForCurrentThread();
}
return workStealingQueue;
}
private WorkStealingQueue CreateWorkStealingQueueForCurrentThread()
{
lock (GlobalQueuesLock)
{
WorkStealingQueue workStealingQueues = _workStealingQueues;
WorkStealingQueue workStealingQueue = ((workStealingQueues != null) ? GetUnownedWorkStealingQueue() : null);
if (workStealingQueue == null)
{
workStealingQueue = (_workStealingQueues = new WorkStealingQueue(workStealingQueues));
}
_locals.Value = workStealingQueue;
return workStealingQueue;
}
}
private WorkStealingQueue GetUnownedWorkStealingQueue()
{
int currentManagedThreadId = Environment.CurrentManagedThreadId;
for (WorkStealingQueue workStealingQueue = _workStealingQueues; workStealingQueue != null; workStealingQueue = workStealingQueue._nextQueue)
{
if (workStealingQueue._ownerThreadId == currentManagedThreadId)
{
return workStealingQueue;
}
}
return null;
}
private bool TrySteal([MaybeNullWhen(false)] out T result, bool take)
{
if (CDSCollectionETWBCLProvider.Log.IsEnabled())
{
if (take)
{
CDSCollectionETWBCLProvider.Log.ConcurrentBag_TryTakeSteals();
}
else
{
CDSCollectionETWBCLProvider.Log.ConcurrentBag_TryPeekSteals();
}
}
while (true)
{
long num = Interlocked.Read(ref _emptyToNonEmptyListTransitionCount);
WorkStealingQueue currentThreadWorkStealingQueue = GetCurrentThreadWorkStealingQueue(forceCreate: false);
bool num2;
if (currentThreadWorkStealingQueue != null)
{
if (TryStealFromTo(currentThreadWorkStealingQueue._nextQueue, null, out result, take))
{
goto IL_0078;
}
num2 = TryStealFromTo(_workStealingQueues, currentThreadWorkStealingQueue, out result, take);
}
else
{
num2 = TryStealFromTo(_workStealingQueues, null, out result, take);
}
if (!num2)
{
if (Interlocked.Read(ref _emptyToNonEmptyListTransitionCount) == num)
{
break;
}
continue;
}
goto IL_0078;
IL_0078:
return true;
}
return false;
}
private bool TryStealFromTo(WorkStealingQueue startInclusive, WorkStealingQueue endExclusive, [MaybeNullWhen(false)] out T result, bool take)
{
for (WorkStealingQueue workStealingQueue = startInclusive; workStealingQueue != endExclusive; workStealingQueue = workStealingQueue._nextQueue)
{
if (workStealingQueue.TrySteal(out result, take))
{
return true;
}
}
result = default(T);
return false;
}
public void CopyTo(T[] array, int index)
{
if (array == null)
{
throw new ArgumentNullException("array", System.SR.ConcurrentBag_CopyTo_ArgumentNullException);
}
if (index array.Length - dangerousCount)
{
throw new ArgumentException(System.SR.Collection_CopyTo_TooManyElems, "index");
}
try
{
int num = CopyFromEachQueueToArray(array, index);
}
catch (ArrayTypeMismatchException ex)
{
throw new InvalidCastException(ex.Message, ex);
}
}
finally
{
UnfreezeBag(lockTaken);
}
}
private int CopyFromEachQueueToArray(T[] array, int index)
{
int num = index;
for (WorkStealingQueue workStealingQueue = _workStealingQueues; workStealingQueue != null; workStealingQueue = workStealingQueue._nextQueue)
{
num += workStealingQueue.DangerousCopyTo(array, num);
}
return num - index;
}
void ICollection.CopyTo(Array array, int index)
{
if (array is T[] array2)
{
CopyTo(array2, index);
return;
}
if (array == null)
{
throw new ArgumentNullException("array", System.SR.ConcurrentBag_CopyTo_ArgumentNullException);
}
ToArray().CopyTo(array, index);
}
public T[] ToArray()
{
if (_workStealingQueues != null)
{
bool lockTaken = false;
try
{
FreezeBag(ref lockTaken);
int dangerousCount = DangerousCount;
if (dangerousCount > 0)
{
T[] array = new T[dangerousCount];
int num = CopyFromEachQueueToArray(array, 0);
return array;
}
}
finally
{
UnfreezeBag(lockTaken);
}
}
return Array.Empty();
}
public void Clear()
{
if (_workStealingQueues == null)
{
return;
}
WorkStealingQueue currentThreadWorkStealingQueue = GetCurrentThreadWorkStealingQueue(forceCreate: false);
if (currentThreadWorkStealingQueue != null)
{
currentThreadWorkStealingQueue.LocalClear();
if (currentThreadWorkStealingQueue._nextQueue == null && currentThreadWorkStealingQueue == _workStealingQueues)
{
return;
}
}
bool lockTaken = false;
try
{
FreezeBag(ref lockTaken);
for (WorkStealingQueue workStealingQueue = _workStealingQueues; workStealingQueue != null; workStealingQueue = workStealingQueue._nextQueue)
{
T result;
while (workStealingQueue.TrySteal(out result, take: true))
{
}
}
}
finally
{
UnfreezeBag(lockTaken);
}
}
public IEnumerator GetEnumerator()
{
return new Enumerator(ToArray());
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
private void FreezeBag(ref bool lockTaken)
{
Monitor.Enter(GlobalQueuesLock, ref lockTaken);
WorkStealingQueue workStealingQueues = _workStealingQueues;
for (WorkStealingQueue workStealingQueue = workStealingQueues; workStealingQueue != null; workStealingQueue = workStealingQueue._nextQueue)
{
Monitor.Enter(workStealingQueue, ref workStealingQueue._frozen);
}
Interlocked.MemoryBarrier();
for (WorkStealingQueue workStealingQueue2 = workStealingQueues; workStealingQueue2 != null; workStealingQueue2 = workStealingQueue2._nextQueue)
{
if (workStealingQueue2._currentOp != 0)
{
SpinWait spinWait = default(SpinWait);
do
{
spinWait.SpinOnce();
}
while (workStealingQueue2._currentOp != 0);
}
}
}
private void UnfreezeBag(bool lockTaken)
{
if (!lockTaken)
{
return;
}
for (WorkStealingQueue workStealingQueue = _workStealingQueues; workStealingQueue != null; workStealingQueue = workStealingQueue._nextQueue)
{
if (workStealingQueue._frozen)
{
workStealingQueue._frozen = false;
Monitor.Exit(workStealingQueue);
}
}
Monitor.Exit(GlobalQueuesLock);
}
}
2、属性
Count
获取 ConcurrentBag 中包含的元素数
IsEmpty
获取一个值,该值指示 ConcurrentBag 是否为空
3、方法
Add(T)
将对象添加到 ConcurrentBag 中。
Clear()
从 ConcurrentBag 中删除所有值。
CopyTo(T[], Int32)
从指定数组索引开始将 ConcurrentBag 元素复制到现有一维 Array 中。以下示例代码:
ConcurrentBag tempModels = new ConcurrentBag();
tempModels.Add(new TempModel() { Code = "1", Name = "一" });
tempModels.Add(new TempModel() { Code = "2", Name = "二" });
tempModels.Add(new TempModel() { Code = "3", Name = "三" });
TempModel[] temparr = new TempModel[5];
tempModels.CopyTo(temparr, 1);
输出结果为:
TryPeek(T)
尝试从 ConcurrentBag 返回一个对象但不移除该对象。
TryTake(T)
尝试从 ConcurrentBag 中移除和返回一个对象。
ToString()
返回表示当前对象的字符串。测试值:System.Collections.Concurrent.ConcurrentBag`1[Test.ConsoleApp.TempModel]
ToArray()
将 ConcurrentBag 元素复制到新数组。
GetEnumerator()
获取当前时间的枚举器。 调用后不影响集合的任何更新。枚举器可以安全地与读取、写入 ConcurrentBag 同时使用。
GetHashCode()
获取集合的哈希值。
参考:https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.concurrentbag-1?view=net-5.0
C# ConcurrentBag的实现原理
4、List 和 ConcurrentBag 对比
众所周知,List 集合是非线程安全的,所以我们采用并行编程时会发生丢数据的情况。比如我们通过多线程将一千个对象加入 List,我们最终得到的集合中元素数就会小于一千。
如下测试代码,通过多任务对象 Task 实现将一千个对象加入到 List 中,添加了一千次,但实际上最终的 objects.Count() 值为 913,小于 1000。 但如果将集合名称改成 ConcurrentBag,结果就不会丢失,最终为等于 1000。
static void Main(string[] args)
{
try
{
// List objects = new List();
ConcurrentBag objects = new ConcurrentBag();
Task[] tasks = new Task[1000];
for (int i = 0; i
objects.Add(new MyObject() { Name = "1", Threadnum = Thread.GetCurrentProcessorId() }));
}
Task.WaitAll(tasks); // 等待所有任务完成
Console.WriteLine(objects.Count()); // List:913; ConcurrentBag:1000
Console.ReadLine();
}
catch (Exception ex)
{
}
}
public class MyObject
{
public string Name { get; set; }
public int Threadnum { get; set; }
}
二、Parallel 的使用
任务并行库(TPL)支持通过 System.Threading.Tasks.Parallel 类实现数据操作的并行。Parallel.For 或 Parallel.ForEach 编写的循环逻辑与常见的 for 和 foreach 类似,只是增加并行逻辑,来提升效率。TPL 省去了客户端创建线程或列工作项,同时在基本循环中,不需要加锁,TPL 会处理所有低级别的工作。
常用的方法有 Parallel.For、Parallel.ForEach、Parallel.Invoke 等,下面将一一例举。
1、Parallel.For()
1.1 重载一:Parallel.For(Int32, Int32, Action)
// fromInclusive:开始索引(含) toExclusive:结束索引(不含) body:不允许为 null
public static System.Threading.Tasks.ParallelLoopResult For (int fromInclusive, int toExclusive, Action body);
以下示例使用 For 方法调用 100 个委托,该委托生成随机 Byte 值,并计算其总和:
ParallelLoopResult result = Parallel.For(0, 100,
ctr =>
{
//Random rnd = new Random(ctr * 100000); // public Random(int Seed); // 随机数的种子,若种子相同,多次生成的随机数序列值相同
Random rnd = new Random();
Byte[] bytes = new Byte[100]; // Byte 数组,每个值的范围为 0~255
rnd.NextBytes(bytes); // 生成 100 个 Byte 数值
int sum = 0;
foreach (var byt in bytes) // 再将生成的 100 个数值相加
sum += byt;
Console.WriteLine("Iteration {0,2}: {1:N0}", ctr, sum);
});
Console.WriteLine("Result: Completed Normally");
1.2 比较执行效率 Parallel.For() 和 for()
Paraller.For() 方法类似于 for 循环语句,也是根据入参多次执行同一逻辑操作。使用 Paraller.For() 方法,可以无序的并行运行迭代,而 for 循环只能根据既定的顺序串行运行。
如下实例,对比 Parallel.For() 和 for() 循环的执行效率进行比较:
// 进行 5 此对比
for (int j = 1; j bag = new ConcurrentBag();
var watch = Stopwatch.StartNew();
watch.Start();
for (int i = 0; i ();
watch = Stopwatch.StartNew();
watch.Start();
Parallel.For(0, 20000000, i => // i 为整数序列号
{
bag.Add(i);
});
watch.Stop();
Console.WriteLine($"并行计算:集合有:{bag.Count},总共耗时:{watch.ElapsedMilliseconds}");
}
代码总共执行了五次对比,如下图中的耗时比较,很明显,采用并行的 Parallel.For() 远比串行的 for() 效率要高许多。
1.3 重载二:Parallel.For(Int32, Int32, Action)
// fromInclusive:开始索引(含) toExclusive:结束索引(不含) body:不允许为 null
public static ParallelLoopResult For (int fromInclusive, int toExclusive, Action body);
此重载增加了 System.Threading.Tasks.ParallelLoopState 循环状态参数,从而使得我们可以通过循环状态来控制并行循环的运行。
以下实例,执行 100 次迭代,在随机数 breakIndex 指示的一次迭代时进行中断操作,调用完 Break() 方法后,循环状态的 ShouldExitCurrentIteration 属性值就是 true,然后进入判断if (state.LowestBreakIteration ,当当前迭代序号大于中断时的序号,就直接返回,不再进行后续操作。
var rnd = new Random();
int breakIndex = rnd.Next(1, 11);
Console.WriteLine($"Will call Break at iteration {breakIndex}n");
var result = Parallel.For(1, 101, (i, state) => // 实际执行的是 1 ~ 100,不包含 101
{
Console.WriteLine($"Beginning iteration {i} {Thread.GetCurrentProcessorId()}");
int delay;
lock (rnd)
delay = rnd.Next(1, 1001);
Thread.Sleep(delay);
if (state.ShouldExitCurrentIteration)
{
if (state.LowestBreakIteration
如下是当索引值为 9 时的处理过程:(当迭代序号为 9 时,执行 Break(),此之前已经开始迭代执行的大于 9 的迭代,均直接退出,只有开始没有结束)
1.4 重载三:Parallel.For(Int32, Int32, ParallelOptions, Action)
// fromInclusive:开始索引(含) toExclusive:结束索引(不含) body:不允许为 null
public static ParallelLoopResult For (int fromInclusive, int toExclusive, ParallelOptions parallelOptions, Action body);
此重载在执行 for 循环时,可以配置循环选项 ParallelOptions。
下边是一个实例,通过配置 ParallelOptions 的 CancellationToken 属性,使得循环支持手动取消:
static void Main(string[] args)
{
CancellationTokenSource cancellationSource = new CancellationTokenSource();
ParallelOptions options = new ParallelOptions();
options.CancellationToken = cancellationSource.Token;
try
{
ParallelLoopResult loopResult = Parallel.For( 0, 10, options,
(i, loopState) =>
{
Console.WriteLine("Start Thread={0}, i={1}", Thread.CurrentThread.ManagedThreadId, i);
if (i == 5) // 模拟某次迭代执行时,取消循环
{
cancellationSource.Cancel();
}
for (int j = 0; j // 如果想往上级抛,需要使用 Handle 方法处理一下
//{
// if (p.InnerException.Message == "my god!Exception from childTask1 happend!")
// return true;
// else
// return false; // 返回 false 表示往上继续抛出异常
//});
}
catch (OperationCanceledException ocex) // 专门用于取消循环异常的捕捉
{
Console.WriteLine($"An iteration has triggered a cancellation. THIS WAS EXPECTED.n{ocex}");
}
finally
{
cancellationSource.Dispose();
}
}
如下图中的输出,所有迭代任务都未完成,主要是因为耗时操作执行完成之前,循环就取消了,在if (loopState.ShouldExitCurrentIteration)
判断时,均为 true 就直接返回了。
1.5 重载四:For(Int32, Int32, ParallelOptions, Func, Func, Action)
public static ParallelLoopResult For (int fromInclusive, int toExclusive,
ParallelOptions parallelOptions,
Func localInit,
Func body,
Action localFinally);
以下示例使用线程局部变量来计算许多冗长操作的结果之和。 此示例将并行度限制为 4。
static void Main(string[] args)
{
int result = 0;
int N = 1000000;
Parallel.For(
0, N,
// 限制最多 4 个并行任务
new ParallelOptions { MaxDegreeOfParallelism = 4 },
// Func 初始化本地变量,本地变量是线程独立变量
() => 0,
// Func 迭代操作
(i, loop, localState) =>
{
for (int ii = 0; ii
Interlocked.Add(ref result, localState)
);
Console.WriteLine("实际运算结果: {0}. 目标值: 1000000", result);
Console.ReadLine();
}
如下图输出结果:
参考:https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.for?view=net-7.0
关于 ParallelOptions 详见:https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.paralleloptions?view=net-7.0
2、Parallel.ForEach()
2.1 重载一:Parallel.ForEach(IEnumerable, Action)
public static ParallelLoopResult ForEach (IEnumerable source, Action body);
执行 ForEach 操作,在处理关于 IEnumerable 集合的任务时,可并行运行迭代。
如下代码块,简单的将一个整数数组,输出到控制台:
static void Main(string[] args)
{
int[] ints = { 11, 12, 13, 14, 15, 16, 17, 18, 19 };
ParallelLoopResult result = Parallel.ForEach(ints,
i =>
{
Console.WriteLine(i);
});
Console.ReadLine();
}
从输出结果看,ForEach 操作是无序的:
2.2 重载二:ForEach(IEnumerable, ParallelOptions, Action)
public static ParallelLoopResult ForEach (IEnumerable source, ParallelOptions parallelOptions, Action body);
执行具有 64 位索引(标识待循环集合的顺序)的 foreach 操作,其中在 IEnumerable 上可能会并行运行迭代,而且可以配置循环选项,可以监视和操作循环的状态。
如下示例代码,设置并行任务数为 5,在索引为 6 的任务执行过程中中断循环,看下输出结果:
static void Main(string[] args)
{
// 创建一个集合,其中包含一些数字
var numbers = new int[] { 10, 11, 12, 13, 14, 15, 16, 17, 18, 19 };
// 使用 ParallelOptions 选项设置并行处理的行为
var parallelOptions = new ParallelOptions
{
MaxDegreeOfParallelism = 5
};
Parallel.ForEach(numbers, parallelOptions, (source, loopState, index) => // index:集合中对象的从 0 开始的序号
{
// 在此处编写并行处理逻辑
Console.WriteLine($"开始--Index: {index}, Value: {source}, ThreadId: {Thread.GetCurrentProcessorId()}");
if (loopState.ShouldExitCurrentIteration)
return;
Thread.Sleep(200);
if (index == 6)
loopState.Break();
Console.WriteLine($"结束++Index: {index}, Value: {source}, ThreadId: {Thread.GetCurrentProcessorId()}");
});
Console.ReadLine();
}
如下图输出结果,一次性开始 5 个并行任务,当第 6 个任务进入时,中断循环。
由于操作是无序的,所以在中断之前可能索引在 6 之后的已经开始或者已经执行完成,如下图 8、9 已经执行完毕,7尚未执行。
注意,若允许并行的任务数少时,可能 6 之后的任务都还没来得及开始,另外,每次执行的结果不同。
2.3 重载三:Parallel.ForEach(Partitioner, Action)
public static ParallelLoopResult ForEach (System.Collections.Concurrent.Partitioner source, Action body);
此重载的独到之处,就是可以将数据进行分区,每一个小区内实现串行计算,分区采用 Partitioner.Create() 实现。
long sum = 0;
long sumtop = 10000000;
Stopwatch sw = Stopwatch.StartNew();
Parallel.ForEach(Partitioner.Create(0, sumtop), (range) =>
{
long local = 0;
for (long i = range.Item1; i
关于分区的创建方法 Partitioner.Create(0, Int64)
- 指定了分区的范围,就是 0 ~ Int64;
- 参数中并没有指定分多少个区,默认是系统自动判断执行的。
- 还可以指定分区,做法就是
Partitioner.Create(0, 3000000, Environment.ProcessorCount)
,其中 Environment.ProcessorCount 参数,就对应当前计算机逻辑处理器的数量。
2.4 重载四:ForEach(IEnumerable, Func, Func, Action)
执行具有线程本地数据的 foreach 操作,其中在 IEnumerable 上可能会并行运行迭代,而且可以监视和操作循环的状态。
public static ParallelLoopResult ForEach (IEnumerable source,
Func localInit,
Func body,
Action localFinally);
如下示例,将全部整数逐个输出并且最后在输出他们之和:
static void Main(string[] args)
{
// 全部值的和为 40
int[] input = { 4, 1, 6, 2, 9, 5, 10, 3 };
int sum = 0;
try
{
Parallel.ForEach(
// IEnumerable 可枚举的数据源
input,
// Func 用于返回每个任务的【本地数据的初始状态】的函数委托
// 本示例中的目的就是将 TLocal localSum 的值在每次迭代都赋值为 0
() => 0,
// Func 将为每个迭代调用一次的委托
(n, loopState, localSum) =>
{
localSum += n;
Console.WriteLine($"Thread={Thread.CurrentThread.ManagedThreadId}, n={n}, localSum={localSum}");
return localSum;
},
// Action 用于对每个任务的本地状态执行一个最终操作的委托
// 此示例中的作用是将每个值逐一求和,并返回 sum
(localSum) =>
Interlocked.Add(ref sum, localSum)
);
Console.WriteLine("nSum={0}", sum);
}
catch (AggregateException e)
{
Console.WriteLine("Parallel.ForEach has thrown an exception. This was not expected.n{0}", e);
}
Console.ReadLine();
}
如下输出结果,其中 localSum 在每个线程中初始值都是 0,在其他线程中参与的求和运算,不影响当前线程。
2.5 比较执行效率 for、Parallel.For()、Parallel.For()+TLocal、Parallel.ForEach(Partitioner.Create(), Action)
static void Main(string[] args)
{
Stopwatch sw = null;
long sum = 0;
long sumtop = 10000000;
// 常规 for 循环
sw = Stopwatch.StartNew();
for (long i = 0; i Interlocked.Add(ref sum, item));
sw.Stop();
Console.WriteLine($"result = {sum}, time = {sw.ElapsedMilliseconds} ms --Parallel.For() 方式");
// Parallel.For() + TLocal
sum = 0;
sw = Stopwatch.StartNew();
Parallel.For(
0L, sumtop,
() => 0L,
(item, state, prevLocal) =>
prevLocal + item,
local =>
Interlocked.Add(ref sum, local));
sw.Stop();
Console.WriteLine($"result = {sum}, time = {sw.ElapsedMilliseconds} ms --Parallel.For() + locals 方式");
// Partitioner.Create() 分区方式
sum = 0;
sw = Stopwatch.StartNew();
Parallel.ForEach(Partitioner.Create(0L, sumtop), (range) =>
{
long local = 0;
for (long i = range.Item1; i
如下输出结果,效率最高的显然是自动分区的方式,比常规的 for 循环块将近一倍。最慢的是 Parallel.For() 方式,由于加锁求和导致上下文频繁切换比较耗时,因此这种求和的计算模式不适用。
参考:https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.foreach?view=net-7.0
3、Parallel.ForEachAsync()
Parallel.ForEachAsync() 是在 .NET 6 中新增的一个 API,是 Parallel.ForEach() 的异步版本。https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-7.0
下面简单说明一下 Parallel.ForEach() 和 Parallel.ForEachAsync() 的区别。
- Parallel.ForEach() 是在默认多个或指定的个数的线程下执行的。而 Parallel.ForEachAsync() 不一定是多线程的,强调的是异步而已。
- 若目标集合必须按照顺序执行,则不能选用 Parallel.ForEach() 方法,因为它是无序执行的。
- 当待处理的数据量很大或者执行过程比较耗时,则选用多线程执行的 Parallel.ForEach() 方法更好。
下面是一个关于重载 ForEachAsync(IAsyncEnumerable, ParallelOptions, Func) 的一个简单示例代码:
static async Task Main(string[] args)
{
var nums = Enumerable.Range(0, 10).ToArray();
await Parallel.ForEachAsync(
nums,
new ParallelOptions { MaxDegreeOfParallelism = 3 }, // 配置最多同时分配三个线程
async (i, token) => // Func // 其中 ValueTask 提供异步操作的可等待结果,指的是下文 await 的内容
{
Console.WriteLine($"开始迭代任务 {i} ThreadId:{Thread.GetCurrentProcessorId()}");
// public static Task Delay(int millisecondsDelay, CancellationToken cancellationToken)
// 在指定毫秒后,调用 token 取消当前任务
await Task.Delay(1000, token);
Console.WriteLine($"完成迭代任务 {i}");
});
Console.WriteLine("Finished!");
Console.ReadLine();
}
详情可参考:https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-7.0 ; https://www.gregbair.dev/posts/parallel-foreachasync/
4、Parallel.Invoke()
尽可能并行执行提供的每个操作。
4.1 两个重载:Invoke(Action[])、Invoke(ParallelOptions, Action[])
下面是一个运用 Invoke(Action[]) 重载的示例,分别加入了三个操作,然后看执行结果。第二个重载是在第一个重载的基础上加了并行选项 ParallelOptions 就不在赘述了。
static void Main(string[] args)
{
try
{
Parallel.Invoke(
BasicAction, // 第一个操作 - 静态方法
() => // 第二个操作 - 箭头函数
{
Console.WriteLine("Method=beta, Thread={0}", Thread.CurrentThread.ManagedThreadId);
},
delegate () // 第三个操作 - 委托函数
{
Console.WriteLine("Method=gamma, Thread={0}", Thread.CurrentThread.ManagedThreadId);
}
);
}
catch (AggregateException e)
{
Console.WriteLine("An action has thrown an exception. THIS WAS UNEXPECTED.n{0}", e.InnerException.ToString());
}
Console.ReadLine();
}
static void BasicAction()
{
Console.WriteLine("Method=alpha, Thread={0}", Thread.CurrentThread.ManagedThreadId);
}
由输出结果可知,三个操作是无序的、多线程执行的。
两个参考:https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.invoke?view=net-7.0 Parallel的使用
三、简单总结一下下
实际上看的资料再多,如果没用到实际开发当中就是无用功,下边简单总结一下吧。
由本文 1.2 比较执行效率 Parallel.For() 和 for() 中可知:
- 对于大批量耗时且顺序要求不高的场景可以采用 Parallel.For() 方法,如果对次序有依赖,则只能采用常用的 for 循环。
- 对于操作简单的循环操作,Parallel.For() 就不太适合了,因为多线程操作涉及到上下文的切换,过多的切换场景会严重影响程序运行的效率。
由本文 2.5 比较执行效率 for、Parallel.For()、Parallel.For()+TLocal、Parallel.ForEach(Partitioner.Create(), Action) 中可知:
- 由于示例中的操作比较简单,此时 Parallel.For() 上下文的的切换耗时以及加锁的缺点就凸现了,效率最差。
- 使用线程本地变量(TLocal)的 Parallel.For() 可以避免将大量的访问同步为共享状态的开销,所以可以看到效率就高很多。可参考:编写具有线程局部变量的 Parallel.For 循环
- 分区循环操作 Partitioner.Create(0, Int64) 方法的效率最高,因为事先给待处理的任务进行了分区,分区内串行,避免了过多的上下文切换耗时。
注:个人整理,欢迎路过的大佬评论区指正和补充。
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net
机房租用,北京机房租用,IDC机房托管, http://www.fwqtg.net
相关推荐: Vue-CoreVideoPlayer 视频播放器组件
安装 cnpm install -S vue-core-video-player 快速使用 # 在main.js中 import VueCoreVideoPlayer from ‘vue-core-video-player’ Vue.use(VueCoreVi…