c# 异步进阶———— paralel [二]

科技资讯 投稿 6900 0 评论

c# 异步进阶———— paralel [二]

前言

正文

我们在工作中常常使用task await 和 async,也就是将线程池进行了封装,那么还有一些更高级的应用。

static void Main(string[] args
{
	var ints= Enumerable.Range(1, 100;
	var result = Parallel.ForEach(ints, arg =>
	{
		Console.WriteLine(arg;
	};
	
	Console.Read(;
}

可以看到结果是并行的。

public static ParallelLoopResult ForEach<TSource>(IEnumerable<TSource> source, Action<TSource> body
{
	if (source == null
	{
		throw new ArgumentNullException(nameof(source;
	}
	if (body == null
	{
		throw new ArgumentNullException(nameof(body;
	}

	return ForEachWorker<TSource, object>(
		source, s_defaultParallelOptions, body, null, null, null, null, null, null;
}

进行参数检验,然后交给了ForEachWorker。

然后通过不同的类型,进行分类:

进去看就是一个taskreplicator:

public static void Run<TState>(ReplicatableUserAction<TState> action, ParallelOptions options, bool stopOnFirstFailure
{
	int maxConcurrencyLevel = (options.EffectiveMaxConcurrencyLevel > 0 ? options.EffectiveMaxConcurrencyLevel : int.MaxValue;

	TaskReplicator replicator = new TaskReplicator(options, stopOnFirstFailure;
	new Replica<TState>(replicator, maxConcurrencyLevel, CooperativeMultitaskingTaskTimeout_RootTask, action.Start(;

	Replica nextReplica;
	while (replicator._pendingReplicas.TryDequeue(out nextReplica
		nextReplica.Wait(;

	if (replicator._exceptions != null
		throw new AggregateException(replicator._exceptions;
}

    创建了一个taskreplictor,起到管理作用

  1. 然后通过while,让每一个Replica 都运行完毕才推出,达到同步的效果

if (replicator._exceptions != null
	throw new AggregateException(replicator._exceptions;

可以看一下这个,这个是一个比较好的技巧。如果一个运行管理,不用抛出异常,之间在管理中进行运行处理总结。

那么就看下这个start。

protected Replica(TaskReplicator replicator, int maxConcurrency, int timeout
{
	_replicator = replicator;
	_timeout = timeout;
	_remainingConcurrency = maxConcurrency - 1;
	_pendingTask = new Task(s => ((Replicas.Execute(, this;
	_replicator._pendingReplicas.Enqueue(this;
}

public void Start(
{
	_pendingTask.RunSynchronously(_replicator._scheduler;
}

将会运行Execute,是同步的,而不是异步的,也就是说第一个task将会运行在当前线程。

public void Execute(
{
	try
	{
		if (!_replicator._stopReplicating && _remainingConcurrency > 0
		{
			CreateNewReplica(;
			_remainingConcurrency = 0; // new replica is responsible for adding concurrency from now on.
		}

		bool userActionYieldedBeforeCompletion;

		ExecuteAction(out userActionYieldedBeforeCompletion;

		if (userActionYieldedBeforeCompletion
		{
			_pendingTask = new Task(s => ((Replicas.Execute(, this, CancellationToken.None, TaskCreationOptions.None;
			_pendingTask.Start(_replicator._scheduler;
		}
		else
		{
			_replicator._stopReplicating = true;
			_pendingTask = null;
		}
	}
	catch (Exception ex
	{
		LazyInitializer.EnsureInitialized(ref _replicator._exceptions.Enqueue(ex;
		if (_replicator._stopOnFirstFailure
			_replicator._stopReplicating = true;
		_pendingTask = null;
	}
}

一段一段分析:

if (!_replicator._stopReplicating && _remainingConcurrency > 0
{
	CreateNewReplica(;
	_remainingConcurrency = 0; // new replica is responsible for adding concurrency from now on.
}

这里当_replicator 也就是任务复制器没有停止的时候。这里有两种情况会停止,一种是任务完成,一种是任务异常且设置参数异常时候停止。

那么就复制一个副本。

protected override void CreateNewReplica(
{
	Replica<TState> newReplica = new Replica<TState>(_replicator, _remainingConcurrency, GenerateCooperativeMultitaskingTaskTimeout(, _action;
	newReplica._pendingTask.Start(_replicator._scheduler;
}

复制完副本后,那么就开始运行我们的action了。

protected override void ExecuteAction(out bool yieldedBeforeCompletion
{
	_action(ref _state, _timeout, out yieldedBeforeCompletion;
}

这里传入了timeout,这个timeout并不是我们限制我们单个task的运行时间,而是当运行到一定时候后,这个task就停止运行,然后另外启动一个副本。

if (CheckTimeoutReached(loopTimeout
{
	replicationDelegateYieldedBeforeCompletion = true;
	break;
}
if (userActionYieldedBeforeCompletion
{
	_pendingTask = new Task(s => ((Replicas.Execute(, this, CancellationToken.None, TaskCreationOptions.None;
	_pendingTask.Start(_replicator._scheduler;
}
else
{
	_replicator._stopReplicating = true;
	_pendingTask = null;
}

这个是为了符合操作系统的调度思想,跑的越久的,基本上优先级会低些。

while (myPartition.MoveNext(
{
	KeyValuePair<long, TSource> kvp = myPartition.Current;
	long index = kvp.Key;
	TSource value = kvp.Value;

	// Update our iteration index
	if (state != null state.CurrentIteration = index;

	if (simpleBody != null
		simpleBody(value;
	else if (bodyWithState != null
		bodyWithState(value, state;
	else if (bodyWithStateAndIndex != null
		bodyWithStateAndIndex(value, state, index;
	else if (bodyWithStateAndLocal != null
		localValue = bodyWithStateAndLocal(value, state, localValue;
	else
		localValue = bodyWithEverything(value, state, index, localValue;

	if (sharedPStateFlags.ShouldExitLoop(index break;

	// Cooperative multitasking:
	// Check if allowed loop time is exceeded, if so save current state and return.
	// The task replicator will queue up a replacement task. Note that we don't do this on the root task.
	if (CheckTimeoutReached(loopTimeout
	{
		replicationDelegateYieldedBeforeCompletion = true;
		break;
	}
}

就是拉取我们的enumerator的数据,然后simpleBody(value,进行运行我们写的action。

里面的主要功能,Parallel做到了限制副本数,因为我们知道task并不是越多越好。

第三点,Parallel 有一个foreach 进行迭代器的处理,这里不仅仅是让任务可以并行。

static void Main(string[] args
{
	var ints= Enumerable.Range(1, 100;
	var result = Parallel.ForEach(ints,    (arg, state
		=>
	{
		if (state.IsStopped
		{
			return;   
		}
		
		if (arg > 18
		{
			state.Break(;
		}
	};
	if (result.IsCompleted
	{
		Console.WriteLine("完成";
	}
	Console.Read(;
}

可以进行中断。

在上述中,我们知道可以传递一个taskschedule进行,那么这个taskschedule 是干什么的,对我们的任务调度有什么影响呢? 下一节,自我实现taskschedule。

编程笔记 » c# 异步进阶———— paralel [二]

赞同 (36) or 分享 (0)
游客 发表我的评论   换个身份
取消评论

表情
(0)个小伙伴在吐槽