首页 > 编程语言 >C# TPL之Parallel 并行库解密

C# TPL之Parallel 并行库解密

时间：2023-02-20 17:15:00浏览次数：37  
标签：parallelOptions C# TPL 线程 Action new null Parallel

Parallel.For、Parallel.Foreach 的要求：

  1. 同样的数据类型，例如：List<T>,Dictionary<T,F>,IEnumerable<T>,等等集合类的操作
  2. 执行相同的函数：在Parallel.For或者Parallel.Foreach中只能传入一个Action,也就是说对应的数据需要完成的函数操作
  3. Parallel会等待指定数据集合中所有数据执行完相应的函数

Parallel.Foreach 代码样例：

Parallel.ForEach(queryGrayTenantIds, new ParallelOptions() { MaxDegreeOfParallelism = 4 }, tenantId =>
{
     var tempDic = TalentPortalProviderGeteway.ToDoProvider.GetUnDoUserIdsByMultiTenant(tenantId);
     if (tempDic == null。
     {
        return ;
      }
 
       foreach (var pair in tempDic)
       {
          if (todoUserIdAndTenantIdDic.ContainsKey(pair.Key))
          {
             continue;
           }
 
           todoUserIdAndTenantIdDic.Add(pair.Key, pair.Value);
        }
});

Parallel.Foreach有很多个重载的方法，可以指定很多个参数，这项不是本文的重点，就不一一列举了。

  • 第一个参数指定数据类型
  • 第二个参数指定并行参数，样例中设置了最大并行度为4
  • 第三个参数为Action,使用指定的数据类型，执行函数运算。

注意：函数体中使用了Dictionary添加函数运算过程中的数据，如果各个数据之间有冲突时，使用线程不安全数据结构在函数中执行数据操作会导致数据不是预想的结果。

 

解密Parallel.Foreach中的实现

public static ParallelLoopResult ForEach<TSource>(IEnumerable<TSource> source,ParallelOptions parallelOptions,Action<TSource> body){
   if (source == null)
     throw new ArgumentNullException(nameof (source));
   if (body == null)
     throw new ArgumentNullException(nameof (body));
   if (parallelOptions == null)
      throw new ArgumentNullException(nameof (parallelOptions));
   return Parallel.ForEachWorker<TSource, object>(source, parallelOptions, body, (Action<TSource, ParallelLoopState>) null, (Action<TSource, ParallelLoopState, long>) null, (Func<TSource, ParallelLoopState, object, object>) null, (Func<TSource, ParallelLoopState, long, object, object>) null, (Func<object>) null, (Action<object>) null);
    }

可以看到使用了内部方法Parallel.ForEachWorker来做操作。

private static ParallelLoopResult ForEachWorker<TSource, TLocal>(
      IEnumerable<TSource> source,
      ParallelOptions parallelOptions,
      Action<TSource> body,
      Action<TSource, ParallelLoopState> bodyWithState,
      Action<TSource, ParallelLoopState, long> bodyWithStateAndIndex,
      Func<TSource, ParallelLoopState, TLocal, TLocal> bodyWithStateAndLocal,
      Func<TSource, ParallelLoopState, long, TLocal, TLocal> bodyWithEverything,
      Func<TLocal> localInit,
      Action<TLocal> localFinally)
    {
      if (parallelOptions.CancellationToken.IsCancellationRequested)
        throw new OperationCanceledException(parallelOptions.CancellationToken);
      switch (source)
      {
        case TSource[] array:
          return Parallel.ForEachWorker<TSource, TLocal>(array, parallelOptions, body, bodyWithState, bodyWithStateAndIndex, bodyWithStateAndLocal, bodyWithEverything, localInit, localFinally);
        case IList<TSource> list:
          return Parallel.ForEachWorker<TSource, TLocal>(list, parallelOptions, body, bodyWithState, bodyWithStateAndIndex, bodyWithStateAndLocal, bodyWithEverything, localInit, localFinally);
        default:
          return Parallel.PartitionerForEachWorker<TSource, TLocal>((Partitioner<TSource>) Partitioner.Create<TSource>(source), parallelOptions, body, bodyWithState, bodyWithStateAndIndex, bodyWithStateAndLocal, bodyWithEverything, localInit, localFinally);
      }
    }

这里区分了数组和列表和使用分区器执行Foreach，这里为什么要区分这个数据类型呢？

 

数组执行的ForEachWorker

private static ParallelLoopResult ForEachWorker<TSource, TLocal>(
      TSource[] array,
      ParallelOptions parallelOptions,
      Action<TSource> body,
      Action<TSource, ParallelLoopState> bodyWithState,
      Action<TSource, ParallelLoopState, long> bodyWithStateAndIndex,
      Func<TSource, ParallelLoopState, TLocal, TLocal> bodyWithStateAndLocal,
      Func<TSource, ParallelLoopState, long, TLocal, TLocal> bodyWithEverything,
      Func<TLocal> localInit,
      Action<TLocal> localFinally)
    {
      int lowerBound = array.GetLowerBound(0);
      int toExclusive = array.GetUpperBound(0) + 1;
      if (body != null)
        return Parallel.ForWorker<object>(lowerBound, toExclusive, parallelOptions, (Action<int>) (i => body(array[i])), (Action<int, ParallelLoopState>) null, (Func<int, ParallelLoopState, object, object>) null, (Func<object>) null, (Action<object>) null);
      if (bodyWithState != null)
        return Parallel.ForWorker<object>(lowerBound, toExclusive, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) ((i, state) => bodyWithState(array[i], state)), (Func<int, ParallelLoopState, object, object>) null, (Func<object>) null, (Action<object>) null);
      if (bodyWithStateAndIndex != null)
        return Parallel.ForWorker<object>(lowerBound, toExclusive, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) ((i, state) => bodyWithStateAndIndex(array[i], state, (long) i)), (Func<int, ParallelLoopState, object, object>) null, (Func<object>) null, (Action<object>) null);
      return bodyWithStateAndLocal != null ? Parallel.ForWorker<TLocal>(lowerBound, toExclusive, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) null, (Func<int, ParallelLoopState, TLocal, TLocal>) ((i, state, local) => bodyWithStateAndLocal(array[i], state, local)), localInit, localFinally) : Parallel.ForWorker<TLocal>(lowerBound, toExclusive, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) null, (Func<int, ParallelLoopState, TLocal, TLocal>) ((i, state, local) => bodyWithEverything(array[i], state, (long) i, local)), localInit, localFinally);
    }

 

列表执行的ForEachWorker

private static ParallelLoopResult ForEachWorker<TSource, TLocal>(
      IList<TSource> list,
      ParallelOptions parallelOptions,
      Action<TSource> body,
      Action<TSource, ParallelLoopState> bodyWithState,
      Action<TSource, ParallelLoopState, long> bodyWithStateAndIndex,
      Func<TSource, ParallelLoopState, TLocal, TLocal> bodyWithStateAndLocal,
      Func<TSource, ParallelLoopState, long, TLocal, TLocal> bodyWithEverything,
      Func<TLocal> localInit,
      Action<TLocal> localFinally)
    {
      if (body != null)
        return Parallel.ForWorker<object>(0, list.Count, parallelOptions, (Action<int>) (i => body(list[i])), (Action<int, ParallelLoopState>) null, (Func<int, ParallelLoopState, object, object>) null, (Func<object>) null, (Action<object>) null);
      if (bodyWithState != null)
        return Parallel.ForWorker<object>(0, list.Count, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) ((i, state) => bodyWithState(list[i], state)), (Func<int, ParallelLoopState, object, object>) null, (Func<object>) null, (Action<object>) null);
      if (bodyWithStateAndIndex != null)
        return Parallel.ForWorker<object>(0, list.Count, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) ((i, state) => bodyWithStateAndIndex(list[i], state, (long) i)), (Func<int, ParallelLoopState, object, object>) null, (Func<object>) null, (Action<object>) null);
      return bodyWithStateAndLocal != null ? Parallel.ForWorker<TLocal>(0, list.Count, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) null, (Func<int, ParallelLoopState, TLocal, TLocal>) ((i, state, local) => bodyWithStateAndLocal(list[i], state, local)), localInit, localFinally) : Parallel.ForWorker<TLocal>(0, list.Count, parallelOptions, (Action<int>) null, (Action<int, ParallelLoopState>) null, (Func<int, ParallelLoopState, TLocal, TLocal>) ((i, state, local) => bodyWithEverything(list[i], state, (long) i, local)), localInit, localFinally);
    }

分区器执行的ForEachWorker

private static ParallelLoopResult PartitionerForEachWorker<TSource, TLocal>(
      Partitioner<TSource> source,
      ParallelOptions parallelOptions,
      Action<TSource> simpleBody,
      Action<TSource, ParallelLoopState> bodyWithState,
      Action<TSource, ParallelLoopState, long> bodyWithStateAndIndex,
      Func<TSource, ParallelLoopState, TLocal, TLocal> bodyWithStateAndLocal,
      Func<TSource, ParallelLoopState, long, TLocal, TLocal> bodyWithEverything,
      Func<TLocal> localInit,
      Action<TLocal> localFinally)
    {
      OrderablePartitioner<TSource> orderedSource = source as OrderablePartitioner<TSource>;
      if (!source.SupportsDynamicPartitions)
        throw new InvalidOperationException(Environment.GetResourceString("Parallel_ForEach_PartitionerNotDynamic"));
      CancellationToken cancellationToken = parallelOptions.CancellationToken;
      if (cancellationToken.IsCancellationRequested)
        throw new OperationCanceledException(parallelOptions.CancellationToken);
      int forkJoinContextID = 0;
      Task task = (Task) null;
      if (TplEtwProvider.Log.IsEnabled())
      {
        forkJoinContextID = Interlocked.Increment(ref Parallel.s_forkJoinContextID);
        task = Task.InternalCurrent;
        TplEtwProvider.Log.ParallelLoopBegin(task != null ? task.m_taskScheduler.Id : TaskScheduler.Current.Id, task != null ? task.Id : 0, forkJoinContextID, TplEtwProvider.ForkJoinOperationType.ParallelForEach, 0L, 0L);
      }
      ParallelLoopStateFlags64 sharedPStateFlags = new ParallelLoopStateFlags64();
      ParallelLoopResult parallelLoopResult = new ParallelLoopResult();
      OperationCanceledException oce = (OperationCanceledException) null;
      CancellationTokenRegistration tokenRegistration = new CancellationTokenRegistration();
      cancellationToken = parallelOptions.CancellationToken;
      if (cancellationToken.CanBeCanceled)
      {
        cancellationToken = parallelOptions.CancellationToken;
        tokenRegistration = cancellationToken.InternalRegisterWithoutEC((Action<object>) (o =>
        {
          sharedPStateFlags.Cancel();
          oce = new OperationCanceledException(parallelOptions.CancellationToken);
        }), (object) null);
      }
      IEnumerable<TSource> partitionerSource = (IEnumerable<TSource>) null;
      IEnumerable<KeyValuePair<long, TSource>> orderablePartitionerSource = (IEnumerable<KeyValuePair<long, TSource>>) null;
      if (orderedSource != null)
      {
        //使用系统默认分区器获取分区数据
        orderablePartitionerSource = orderedSource.GetOrderableDynamicPartitions();
        if (orderablePartitionerSource == null)
          throw new InvalidOperationException(Environment.GetResourceString("Parallel_ForEach_PartitionerReturnedNull"));
      }
      else
      {
        partitionerSource = source.GetDynamicPartitions();
        if (partitionerSource == null)
          throw new InvalidOperationException(Environment.GetResourceString("Parallel_ForEach_PartitionerReturnedNull"));
      }
      //可复制任务，当前Parallel中的根任务
      ParallelForReplicatingTask rootTask = (ParallelForReplicatingTask) null;
      //可复制任务执行的函数
      Action action = (Action) (() =>
      {
        Task internalCurrent = Task.InternalCurrent;
        if (TplEtwProvider.Log.IsEnabled())
          TplEtwProvider.Log.ParallelFork(internalCurrent != null ? internalCurrent.m_taskScheduler.Id : TaskScheduler.Current.Id, internalCurrent != null ? internalCurrent.Id : 0, forkJoinContextID);
        TLocal local = default (TLocal);
        bool flag1 = false;
        IDisposable disposable = (IDisposable) null;
        try
        {
          ParallelLoopState64 parallelLoopState64 = (ParallelLoopState64) null;
          if (bodyWithState != null || bodyWithStateAndIndex != null)
            parallelLoopState64 = new ParallelLoopState64(sharedPStateFlags);
          else if (bodyWithStateAndLocal != null || bodyWithEverything != null)
          {
            parallelLoopState64 = new ParallelLoopState64(sharedPStateFlags);
            if (localInit != null)
            {
              local = localInit();
              flag1 = true;
            }
          }
          bool flag2 = rootTask == internalCurrent; //初始化可复制任务
          Parallel.LoopTimer loopTimer = new Parallel.LoopTimer(rootTask.ActiveChildCount); //并行处理器计时器
          if (orderedSource != null)
          {
            if (!(internalCurrent.SavedStateFromPreviousReplica is IEnumerator<KeyValuePair<long, TSource>> enumerator3))
            {
              enumerator3 = orderablePartitionerSource.GetEnumerator(); //获取分区器的循环迭代指针
              if (enumerator3 == null)
                throw new InvalidOperationException(Environment.GetResourceString("Parallel_ForEach_NullEnumerator"));
            }
            disposable = (IDisposable) enumerator3;
            while (enumerator3.MoveNext()) //循环迭代指针
            {
              KeyValuePair<long, TSource> current = enumerator3.Current;
              long key = current.Key;
              TSource source1 = current.Value;
              if (parallelLoopState64 != null)
                parallelLoopState64.CurrentIteration = key;
              if (simpleBody != null)
                simpleBody(source1); //执行Parallel中指定的Action
              else if (bodyWithState != null)
                bodyWithState(source1, (ParallelLoopState) parallelLoopState64); //执行Parallel中指定的Action
              else if (bodyWithStateAndIndex != null)
                bodyWithStateAndIndex(source1, (ParallelLoopState) parallelLoopState64, key); //执行Parallel中指定的Action
              else
                local = bodyWithStateAndLocal == null ? bodyWithEverything(source1, (ParallelLoopState) parallelLoopState64, key, local) : bodyWithStateAndLocal(source1, (ParallelLoopState) parallelLoopState64, local);
              if (sharedPStateFlags.ShouldExitLoop(key))
                break;
              if (!flag2 && loopTimer.LimitExceeded()) //是否到达循环器限制
              {
                internalCurrent.SavedStateForNextReplica = (object) enumerator3;
                disposable = (IDisposable) null;
                break; //停止循环
              }
            }
          }
          else  //自定义分区器
          {
            if (!(internalCurrent.SavedStateFromPreviousReplica is IEnumerator<TSource> enumerator4))
            {
              enumerator4 = partitionerSource.GetEnumerator();
              if (enumerator4 == null)
                throw new InvalidOperationException(Environment.GetResourceString("Parallel_ForEach_NullEnumerator"));
            }
            disposable = (IDisposable) enumerator4;
            if (parallelLoopState64 != null)
              parallelLoopState64.CurrentIteration = 0L;
            while (enumerator4.MoveNext())
            {
              TSource current = enumerator4.Current;
              if (simpleBody != null)
                simpleBody(current);
              else if (bodyWithState != null)
                bodyWithState(current, (ParallelLoopState) parallelLoopState64);
              else if (bodyWithStateAndLocal != null)
                local = bodyWithStateAndLocal(current, (ParallelLoopState) parallelLoopState64, local);
              if (sharedPStateFlags.LoopStateFlags != ParallelLoopStateFlags.PLS_NONE)
                break;
              if (!flag2 && loopTimer.LimitExceeded())
              {
                internalCurrent.SavedStateForNextReplica = (object) enumerator4;
                disposable = (IDisposable) null;
                break;
              }
            }
          }
        }
        catch
        {
          sharedPStateFlags.SetExceptional();
          throw;
        }
        finally
        {
          if (localFinally != null & flag1)
            localFinally(local);
          disposable?.Dispose();
          if (TplEtwProvider.Log.IsEnabled())
            TplEtwProvider.Log.ParallelJoin(internalCurrent != null ? internalCurrent.m_taskScheduler.Id : TaskScheduler.Current.Id, internalCurrent != null ? internalCurrent.Id : 0, forkJoinContextID);
        }
      });
      try
      {
        rootTask = new ParallelForReplicatingTask(parallelOptions, action, TaskCreationOptions.None, InternalTaskOptions.SelfReplicating);  //初始化根任务
        rootTask.RunSynchronously(parallelOptions.EffectiveTaskScheduler); //运行异步任务，使用当前指定的任务调度器
        rootTask.Wait();//等待所有任务执行完毕
        cancellationToken = parallelOptions.CancellationToken;
        if (cancellationToken.CanBeCanceled)
          tokenRegistration.Dispose();
        if (oce != null)
          throw oce;
      }
      catch (AggregateException ex)
      {
        if (parallelOptions.CancellationToken.CanBeCanceled)
          tokenRegistration.Dispose();
        Parallel.ThrowIfReducableToSingleOCE((IEnumerable<Exception>) ex.InnerExceptions, parallelOptions.CancellationToken);
        throw;
      }
      catch (TaskSchedulerException ex)
      {
        if (parallelOptions.CancellationToken.CanBeCanceled)
          tokenRegistration.Dispose();
        throw;
      }
      finally
      {
        int loopStateFlags = sharedPStateFlags.LoopStateFlags;
        parallelLoopResult.m_completed = loopStateFlags == ParallelLoopStateFlags.PLS_NONE;
        if ((loopStateFlags & ParallelLoopStateFlags.PLS_BROKEN) != 0)
          parallelLoopResult.m_lowestBreakIteration = new long?(sharedPStateFlags.LowestBreakIteration);
        if (rootTask != null && rootTask.IsCompleted)
          rootTask.Dispose();
        (orderablePartitionerSource == null ? partitionerSource as IDisposable : orderablePartitionerSource as IDisposable)?.Dispose();
        if (TplEtwProvider.Log.IsEnabled())
          TplEtwProvider.Log.ParallelLoopEnd(task != null ? task.m_taskScheduler.Id : TaskScheduler.Current.Id, task != null ? task.Id : 0, forkJoinContextID, 0L);
      }
      return parallelLoopResult;
    }

 

数组和列表执行Parallel.Foreach的区别

这里可以看出区别，数组和列表的不同是在于获取开始位置和结束位置，数组使用的方法是外部扩展方法，其实这个方法在C语言中比较常见。

GetLowerBound 获取数组中第一维度为0的下限。

GetUpperBound 获取数组中第一维度0的上限。
举例：

string[,] Arr= new string[,]{{"1","2"},{"3","4"},{"5","6"}};
for(int i=Arr.GetLowerBound(0);i<=Arr.GetUpperBound(0);i++)
{
 
//Arr.GetLowerBound(0);其中的0表示取第一维的下限，一般数组索引是0开始，为0
 
//同理可得Arr.GetUpperBound(0);其中的0表示取第一维的上限，在本例中是3行2列的数组，所以为3-1=2
}

列表和数组最终都会执行到同一个方法

private static ParallelLoopResult ForWorker<TLocal>(
      int fromInclusive,
      int toExclusive,
      ParallelOptions parallelOptions,
      Action<int> body,
      Action<int, ParallelLoopState> bodyWithState,
      Func<int, ParallelLoopState, TLocal, TLocal> bodyWithLocal,
      Func<TLocal> localInit,
      Action<TLocal> localFinally)
    {
      ParallelLoopResult parallelLoopResult = new ParallelLoopResult();
      if (toExclusive <= fromInclusive)
      {
        parallelLoopResult.m_completed = true;
        return parallelLoopResult;
      }
      ParallelLoopStateFlags32 sharedPStateFlags = new ParallelLoopStateFlags32();
      TaskCreationOptions creationOptions = TaskCreationOptions.None;
      InternalTaskOptions internalOptions = InternalTaskOptions.SelfReplicating;
      if (parallelOptions.CancellationToken.IsCancellationRequested)
        throw new OperationCanceledException(parallelOptions.CancellationToken);
      int nNumExpectedWorkers = parallelOptions.EffectiveMaxConcurrencyLevel == -1 ? PlatformHelper.ProcessorCount : parallelOptions.EffectiveMaxConcurrencyLevel; //获取指定的最大并行度，没有设置获取当前系统的最大任务线程数
      RangeManager rangeManager = new RangeManager((long) fromInclusive, (long) toExclusive, 1L, nNumExpectedWorkers); //随机分区管理器，计算方法为最小下标至最大下标随机分配，最多创建指定最大并行数个分区
      OperationCanceledException oce = (OperationCanceledException) null;
      CancellationTokenRegistration tokenRegistration = new CancellationTokenRegistration();
      CancellationToken cancellationToken;
      if (parallelOptions.CancellationToken.CanBeCanceled)
      {
        cancellationToken = parallelOptions.CancellationToken;
        tokenRegistration = cancellationToken.InternalRegisterWithoutEC((Action<object>) (o =>
        {
          sharedPStateFlags.Cancel();
          oce = new OperationCanceledException(parallelOptions.CancellationToken);
        }), (object) null);
      }
      int forkJoinContextID = 0;
      Task task = (Task) null;
      if (TplEtwProvider.Log.IsEnabled())
      {
        forkJoinContextID = Interlocked.Increment(ref Parallel.s_forkJoinContextID);
        task = Task.InternalCurrent;
        TplEtwProvider.Log.ParallelLoopBegin(task != null ? task.m_taskScheduler.Id : TaskScheduler.Current.Id, task != null ? task.Id : 0, forkJoinContextID, TplEtwProvider.ForkJoinOperationType.ParallelFor, (long) fromInclusive, (long) toExclusive);
      }
      ParallelForReplicatingTask rootTask = (ParallelForReplicatingTask) null;
      try
      {
        //初始化根任务
        rootTask = new ParallelForReplicatingTask(parallelOptions, (Action) (() =>
        {
          Task internalCurrent = Task.InternalCurrent;
          bool flag1 = internalCurrent == rootTask;
          RangeWorker rangeWorker2 = new RangeWorker();
          if (!(internalCurrent.SavedStateFromPreviousReplica is RangeWorker rangeWorker3))
            rangeWorker3 = rangeManager.RegisterNewWorker(); //如果当前任务的上一个任务是根任务，那么重新生成一个分区
          int nFromInclusiveLocal32;
          int nToExclusiveLocal32;
          if (!rangeWorker3.FindNewWork32(out nFromInclusiveLocal32, out nToExclusiveLocal32) || sharedPStateFlags.ShouldExitLoop(nFromInclusiveLocal32)) //当前分区是否已经循环完毕
            return;
          if (TplEtwProvider.Log.IsEnabled())
            TplEtwProvider.Log.ParallelFork(internalCurrent != null ? internalCurrent.m_taskScheduler.Id : TaskScheduler.Current.Id, internalCurrent != null ? internalCurrent.Id : 0, forkJoinContextID);
          TLocal local = default (TLocal);
          bool flag2 = false;
          try
          {
            ParallelLoopState32 parallelLoopState32 = (ParallelLoopState32) null;
            if (bodyWithState != null)
              parallelLoopState32 = new ParallelLoopState32(sharedPStateFlags);
            else if (bodyWithLocal != null)
            {
              parallelLoopState32 = new ParallelLoopState32(sharedPStateFlags);
              if (localInit != null)
              {
                local = localInit();
                flag2 = true;
              }
            }
            Parallel.LoopTimer loopTimer = new Parallel.LoopTimer(rootTask.ActiveChildCount); //并行计时器
            do
            {
              if (body != null)
              {
                for (int index = nFromInclusiveLocal32; index < nToExclusiveLocal32 && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.PLS_NONE || !sharedPStateFlags.ShouldExitLoop()); ++index) //循环分区中的最小下标和最大小，获取集合中的数据，执行指定的Action
                  body(index);
              }
              else if (bodyWithState != null)
              {
                for (int CallerIteration = nFromInclusiveLocal32; CallerIteration < nToExclusiveLocal32 && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.PLS_NONE || !sharedPStateFlags.ShouldExitLoop(CallerIteration)); ++CallerIteration)
                {
                  parallelLoopState32.CurrentIteration = CallerIteration;
                  bodyWithState(CallerIteration, (ParallelLoopState) parallelLoopState32);
                }
              }
              else
              {
                for (int CallerIteration = nFromInclusiveLocal32; CallerIteration < nToExclusiveLocal32 && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.PLS_NONE || !sharedPStateFlags.ShouldExitLoop(CallerIteration)); ++CallerIteration)
                {
                  parallelLoopState32.CurrentIteration = CallerIteration;
                  local = bodyWithLocal(CallerIteration, (ParallelLoopState) parallelLoopState32, local);
                }
              }
              if (!flag1 && loopTimer.LimitExceeded())
              {
                internalCurrent.SavedStateForNextReplica = (object) rangeWorker3;
                break;
              }
            }
            while (rangeWorker3.FindNewWork32(out nFromInclusiveLocal32, out nToExclusiveLocal32) && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.PLS_NONE || !sharedPStateFlags.ShouldExitLoop(nFromInclusiveLocal32)));  //寻找下一个分区，执行下个分区的数据
          }
          catch
          {
            sharedPStateFlags.SetExceptional();
            throw;
          }
          finally
          {
            if (localFinally != null & flag2)
              localFinally(local);
            if (TplEtwProvider.Log.IsEnabled())
              TplEtwProvider.Log.ParallelJoin(internalCurrent != null ? internalCurrent.m_taskScheduler.Id : TaskScheduler.Current.Id, internalCurrent != null ? internalCurrent.Id : 0, forkJoinContextID);
          }
        }), creationOptions, internalOptions);
        rootTask.RunSynchronously(parallelOptions.EffectiveTaskScheduler);  //使用当前任务调度器异步执行任务
        rootTask.Wait(); //等待所有任务完成
        cancellationToken = parallelOptions.CancellationToken;
        if (cancellationToken.CanBeCanceled)
          tokenRegistration.Dispose();
        if (oce != null)
          throw oce;
      }
      catch (AggregateException ex)
      {
        if (parallelOptions.CancellationToken.CanBeCanceled)
          tokenRegistration.Dispose();
        Parallel.ThrowIfReducableToSingleOCE((IEnumerable<Exception>) ex.InnerExceptions, parallelOptions.CancellationToken);
        throw;
      }
      catch (TaskSchedulerException ex)
      {
        if (parallelOptions.CancellationToken.CanBeCanceled)
          tokenRegistration.Dispose();
        throw;
      }
      finally
      {
        int loopStateFlags = sharedPStateFlags.LoopStateFlags;
        parallelLoopResult.m_completed = loopStateFlags == ParallelLoopStateFlags.PLS_NONE;
        if ((loopStateFlags & ParallelLoopStateFlags.PLS_BROKEN) != 0)
          parallelLoopResult.m_lowestBreakIteration = new long?((long) sharedPStateFlags.LowestBreakIteration);
        if (rootTask != null && rootTask.IsCompleted)
          rootTask.Dispose();
        if (TplEtwProvider.Log.IsEnabled())
        {
          int TotalIterations = loopStateFlags != ParallelLoopStateFlags.PLS_NONE ? ((loopStateFlags & ParallelLoopStateFlags.PLS_BROKEN) == 0 ? -1 : sharedPStateFlags.LowestBreakIteration - fromInclusive) : toExclusive - fromInclusive;
          TplEtwProvider.Log.ParallelLoopEnd(task != null ? task.m_taskScheduler.Id : TaskScheduler.Current.Id, task != null ? task.Id : 0, forkJoinContextID, (long) TotalIterations);
        }
      }
      return parallelLoopResult;
    }

通过上述代码观察，Parallel.Foreach的运行核心方法是

  1. 数据分区，使用分区将数据创建多个小的类似集合
  2. 可复制任务，数据分区之后，使用可复制任务执行Action，这样可以将数据分成小份之后，执行多个相同的任务
  3. 同步执行多个任务，等待所有任务执行完成，返回总集合结果

那么核心就是同步任务这个类是如何运行Action的呢？

RootTask解密

RootTask内部方法InternalRunSynchronously

internal void InternalRunSynchronously(TaskScheduler scheduler, bool waitForCompletion)
    {
      int stateFlags = this.m_stateFlags;
      TaskCreationOptions taskCreationOptions = Task.OptionsMethod(stateFlags);
      if ((taskCreationOptions & (TaskCreationOptions) 512) != TaskCreationOptions.None)
        throw new InvalidOperationException(Environment.GetResourceString("Task_RunSynchronously_Continuation"));
      if ((taskCreationOptions & (TaskCreationOptions) 1024) != TaskCreationOptions.None)
        throw new InvalidOperationException(Environment.GetResourceString("Task_RunSynchronously_Promise"));
      if (Task.IsCompletedMethod(stateFlags))
        throw new InvalidOperationException(Environment.GetResourceString("Task_RunSynchronously_TaskCompleted"));
      if (Interlocked.CompareExchange<TaskScheduler>(ref this.m_taskScheduler, scheduler, (TaskScheduler) null) != null)
        throw new InvalidOperationException(Environment.GetResourceString("Task_RunSynchronously_AlreadyStarted"));
      if (!this.MarkStarted())
        throw new InvalidOperationException(Environment.GetResourceString("Task_RunSynchronously_TaskCompleted"));
      bool flag = false;
      try
      {
        if (!scheduler.TryRunInline(this, false))
        {
          scheduler.InternalQueueTask(this);
          flag = true;
        }
        if (!waitForCompletion || this.IsCompleted)
          return;
        this.SpinThenBlockingWait(-1, new CancellationToken());
      }
      catch (System.Exception ex)
      {
        if (!flag && !(ex is ThreadAbortException))
        {
          TaskSchedulerException exceptionObject = new TaskSchedulerException(ex);
          this.AddException((object) exceptionObject);
          this.Finish(false);
          this.m_contingentProperties.m_exceptionsHolder.MarkAsHandled(false);
          throw exceptionObject;
        }
        throw;
      }
    }

可以看到，RootTask继承自Task，在执行时也是使用调度器来完成任务的执行，上述默认使用ThreadPool任务执行器。现在应该对整个运行任务比较清晰了。重新梳理一遍

  1. 创建分区管理器，将集合数据分区
  2. 创建可复制任务，根任务为启动任务，其他任务为关联任务
  3. 可复制任务中循环执行分区数据
  4. 根任务为启动任务，启动后等待其他关联任务完成
  5. 返回所有数据执行完的结果

TaskSchduler简介

当rootTask执行的时候指定了一个任务调度器，一个Task可以有多种运行方式，一般由TaskCreationOptions枚举指定。

  • None = 0  //不指定类型

  • PreferFairness = 1  //公平运行，在使用ThreadPool运行任务时，越早进入线程池中越早被执行

  • LongRunning = 2 //指定任务将是长时间运行的、粗粒度的操作，涉及比细化的系统更少、更大的组件。此任务会告知线程池单独开启一个长任务后端线程执行此任务

  • AttachedToParent = 4 //附加父任务的子任务，与父任务同步执行

  • DenyChildAttach = 8 //指定任何尝试作为附加的子任务执行（即，使用AttachedToParent选项创建）的子任务都无法附加到父任务，会改成作为分离的子任务执行。

  • HideScheduler = 16  //隐藏任务调度器，使用默认的任务调度器。通常都会使用默认的调度器，当父任务使用自定义调度器，子任务期望使用默认调度器时使用

  • RunContinuationsAsynchronously = 64  //强制任务以异步的方式执行

这个需要回溯一下Task执行一个任务的时候是需要有一个任务调度器来规定当前任务如何执行。

TaskSchduler 采用默认的的任务调度器：ThreadPoolTaskScheduler，当一个任务被QueueTask时，此调度器会判断任务是否时长时间运行任务，如果是那么单独开通一个后端线程运行此任务，如果不是使用线程池运行该任务。

单独开通线程运行任务：指定该线程为长时间任务类型线程

 .net中的任务调度器有哪些

线程池任务调度器：ThreadPoolTaskScheduler  （默认调度器）
核心库任务调度器：ConcurrentExclusiveSchedulerPair
UI任务调度器：SynchronizationContextTaskScheduler，并发度为1 （wpf中使用较多，同步上下文任务调度器，不阻塞UI线程）：

 当前线程设置默认的同步上下文SynchronizationContext.SetSynchronizationContext(new SynchronizationContext());就可以正常使用该方法

通过TaskScheduler.FromCurrentSynchronizationContext() 调用SynchronizationContextTaskScheduler

自定义任务调度器，继承TaskScheduler，实现QueueTask、TryExecuteTaskInline核心方法即可。

实战

以下代码一共执行了几秒？线程如何变化？

var data = new List<string>() { "a", "b", "c" };
var loopResult = Parallel.ForEach(data, (str) =>
{
  if (str.Equals("a"))
  {
    Thread.Sleep(1000);
  }
  Thread.Sleep(1000);
 
 Console.WriteLine("字符串为"+str);
});

以下代码会输出什么？怎么修改可以顺序输出（还需要使用Parallel）？

var data = new List<int>() { 1, 2, 3,4,5 };
var loopResult = Parallel.ForEach(data, (str) =>
{
   Console.WriteLine("数字为"+str);
 });

 

浅谈线程池

线程池管理线程，主要负责线程创建和销毁。使用线程池创建的线程都是后端工作线程，也就是不能使用ThreadPool创建UI线程。使用线程池主要为了降低手动创建线程的开销，线程池可以让线程复用处理不同的功能。不过线程池也将线程的创建和销毁给封装了，让之前可操作的线程操作封装成为黑盒，如果你对线程的创建和销毁有严格的要求可以自己创建一个池化的线程数据结构。

线程池示例图

线程池中线程管理

  • ThreadPool.UnsafeQueueCustomWorkItem((IThreadPoolWorkItem) task)) 默认会创建一个线程

  • .net 会为每个进程生成一个线程池。线程池的初始值线程数是 cpu逻辑内核数。后面连续建线程要每间隔500ms新建一个线程。即使突然并发大量任务也是按这个进度进来线程。
  • 当线程1中没有任务时，会先去公共队列中获取任务，公共队列也没有了，会去别的线程中获取任务，所谓的空闲线程优化。
  • 线程池运行任务：
    32位环境中的  workerThreads：1023  默认大小逻辑cpucore数， completionPortThreads：25   默认大小逻辑cpucore数
    64位环境中的  workerThreads：32767 默认大小逻辑cpucore数, completionPortThreads：1000  默认大小逻辑cpucore数

TaskSchduler和ThreadPool的区别

TaskSchduler默认也是使用ThreadPool,TaskSchduler的优化点在于，生成一个线程池时，会创建一个本地队列，直接将任务放入到本地队列中，减少从全局队列中获取任务时的资源竞争。

线程池添加任务

 

ThreadPoolTaskScheduler 添加任务

当任务创建选项不是公平运行时，使用本地队列，如果是公平运行时，使用全局队列。

标签：parallelOptions,C#,TPL,线程,Action,new,null,Parallel
From： https://www.cnblogs.com/thirteenAnimation/p/17138102.html

相关文章

  • 判断JavaScript数据是否为空
    getObjType=(obj)=>{consttype=Object.prototype.toString.call(obj)constarray=type.split(/\s/g)returnarray[1].replace(']','')}is......
  • 表单 下拉框 多选列表 单选 radio 多选 checkbox
      1usingMicrosoft.AspNetCore.Mvc;2usingMicrosoft.AspNetCore.Mvc.Rendering;3usingSystem.ComponentModel.DataAnnotations;45namespaceMvcMov......
  • 项目中TS的配置文件tsconfig.json
    1.快速搭建一个浏览器开发环境建立好文件夹后，打开VSCode，把文件夹拉到编辑器当中，然后打开终端，运行npminit-y,创建package.json文件。生成文件后，我们接着在终端中运行t......
  • 微服务 - Spring Cloud
    @Configuration(proxyBeanMethods=false)标记了@ConfigurationSpring底层会给配置创建cglib动态代理。作用：就是防止每次调用本类的Bean方法而重新创建对象，Bean是默......
  • acwing 318. 划分大理石
      多重背包及优化#include<bits/stdc++.h>usingnamespacestd;constintN=5e5+10;intc[N],S,n,f[N];intsolve(){ inti,j; S=0; memset(f,0,si......
  • SGDR: Stochastic Gradient Descent with Warm Restarts
    SGDR和CLR比较类似,都是给学习率设置周期和变化范围,按照下面的公式让学习率周期性的变化:总的来说,就是在一个周期开始时,将学习率设置为最大值,然后按照余弦函数减小到......
  • TCP编程之网卡信息获取和域名解析
    1.TCP/IP简介TCP/IP协议源于1969年，是针对Internet开发的一种体系结构和协议标准，目的在于解决异种计算机网络的通信问题。使得网络在互联时能为用户提供一种通用、一致的通信......
  • JAVA中使用最广泛的本地缓存？Ehcache的自信从何而来3 —— 本地缓存变身分布式集群缓存
    大家好，又见面了。本文是笔者作为掘金技术社区签约作者的身份输出的缓存专栏系列内容，将会通过系列专题，讲清楚缓存的方方面面。如果感兴趣，欢迎关注以获取后续更新。上......
  • 高可用Nacos服务实战
     环境：centOS7nacos-server-2.2.0.zipkeepalived-2.2.7.tar.gznginx-1.22.1.tar.gz Nginx+keepalived搭建高可用集群Nacos   实现思路：1、安装nacos集群......
  • 【JDBC：数据库连接池，数据库驱动】
    一：什么是JDBC?1.JDBC（JavaDataBaseConnectivity,java数据库连接）是一种用于执行SQL语句的JavaAPI，可以为多种关系数据库提供统一访问，它由一组用Java语言编写的类和接口组成......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99