blocks|key|2364046|text|对于并行，总是存在一个定义“按顺序”的意思的问题。假设您有一个包含100个项目的集合。“一次处理4个订单”(如您所请求的)可能意味着：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2364047|2364048|松散排序:使用4个线程，并按原始集合的顺序发出任务。|ordered-list-item|2364049|2364050|在这种情况下，您可以使用：|2364051|新项目po+=新代码(){+MaxDegreeOfParallelism+=4+}；新项目po，(+ParallelOptions+)ParallelOptions{+//+MaxDegreeOfParallelism+})；|2364052|在任务不均衡的情况下，会很快失去原来的顺序，因为一些线程在重任务上可能会落后，但会按顺序分配任务。|2364053|严格排序:按4个一组的顺序进行处理，如下所示：|unordered-list-item|2364054|2364055|0+1+2+3+4任务_____________________________屏障4+5+6+7+4任务_____________________________屏障等。|2364056|在这种情况下，您可以使用一个屏障：|2364057|Parallel.ForEach(list.AsParallel().AsOrdered()，b=新屏障(4)；代码po+=新ParallelOptions()+{+MaxDegreeOfParallelism+=4+}；ParallelOptions+po，(item)+=>+{+//+ParallelOptions+b.SignalAndWait()；})；|2364058|尽管您必须确保任务的数量是4的倍数，否则屏障不会在单个任务的最后iterations.|2364059|Process+4项时发出信号:您可以创建一个封装原始列表的4项的任务对象，然后像第一种情况那样执行一个简单的Parallel.ForEach+(即每个线程将作为单个任务的一部分顺序处理4项)。这将按顺序发出4个一组的任务，但如果任务花费的时间太长，可能会再次导致一些线程落后。|offset|length|style|CODE|2364060|2364061|entityMap^0|0|0|0|0|0|0|0|0|0|0|0|0|0|1J|G|0|0^^$0|@$1|2|3|4|5|6|7|17|8|@]|9|@]|A|$]]|$1|B|3|-4|5|6|7|18|8|@]|9|@]|A|$]]|$1|C|3|D|5|E|7|19|8|@]|9|@]|A|$]]|$1|F|3|-4|5|6|7|1A|8|@]|9|@]|A|$]]|$1|G|3|H|5|6|7|1B|8|@]|9|@]|A|$]]|$1|I|3|J|5|6|7|1C|8|@]|9|@]|A|$]]|$1|K|3|L|5|6|7|1D|8|@]|9|@]|A|$]]|$1|M|3|N|5|O|7|1E|8|@]|9|@]|A|$]]|$1|P|3|-4|5|6|7|1F|8|@]|9|@]|A|$]]|$1|Q|3|R|5|6|7|1G|8|@]|9|@]|A|$]]|$1|S|3|T|5|6|7|1H|8|@]|9|@]|A|$]]|$1|U|3|V|5|6|7|1I|8|@]|9|@]|A|$]]|$1|W|3|X|5|6|7|1J|8|@]|9|@]|A|$]]|$1|Y|3|Z|5|O|7|1K|8|@$10|1L|11|1M|12|13]]|9|@]|A|$]]|$1|14|3|-4|5|6|7|1N|8|@]|9|@]|A|$]]|$1|15|3|-4|5|6|7|1O|8|@]|9|@]|A|$]]]|16|$]]

With parallelism there is always a problem of defining what "in order" means. Let's say you have a collection of 100 items. Processing them "in order 4 at a time" (as you requested) could mean:

<ol>
<li>Loose ordering: use 4 threads and issue tasks in the order of the original collection.

In this case you can use:

<pre><code>ParallelOptions po = new ParallelOptions() { MaxDegreeOfParallelism = 4 };
Parallel.ForEach(list.AsParallel().AsOrdered(), po,
 (item) =&gt;
 {
 // code
 });
</code></pre>

In case of unbalanced tasks this will quickly lose the original ordering, as some threads may lag behind on heavy tasks, but the tasks will be allocated in order.</li>
<li>Strict ordering: process them in order in groups of 4 like below:

<pre><code> 0 1 2 3 
 4 tasks
 _____________________________
 barrier

 4 5 6 7 
 4 tasks
 _____________________________
 barrier

 etc.
</code></pre>

In this case you can employ a barrier:

<pre><code>Barrier b = new Barrier(4);
ParallelOptions po = new ParallelOptions() { MaxDegreeOfParallelism = 4 };
Parallel.ForEach(list.AsParallel().AsOrdered(), po,
 (item) =&gt;
 {
 // code
 b.SignalAndWait();
 });
</code></pre>

Although you have to make sure that the number of tasks is a multiple of 4, otherwise the barrier won't be signaled at the last iterations.</li>
<li>Process 4 items in a single task: you can create a task object that encapsulates 4 items of the original list and then do a simple <code>Parallel.ForEach</code> like in the first case (i.e. each thread will process 4 items sequentially as a part of a single task). This would issue tasks in groups of 4 in order, but again may cause some threads to lag behind if a task takes too long.</li>
</ol>

blocks|key|4252975|text|我不清楚你到底在做什么，“元素是随机选择的”。但是如果您使用Paralle.ForEach()，那么它会试图提高效率，因此它会以某种方式对输入序列进行分区。如果输入序列是IList<T>，它将使用范围分区，否则，它将使用块分区(请参阅)。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|4252976|如果希望按顺序处理这些项，可以使用custom+partitioner配置Parallel.ForEach()，这会将集合划分为大小为1的块。|4252977|但由于您在这里并不真正需要Parallel.ForEach()，因此可能更简单的解决方案是创建4个任务，逐个处理这些项。对于同步，您可以使用BlockingCollection。类似于：|4252978|public+static+class+ParallelOrdered
{
++++public+static+void+ForEach<T>(IEnumerable<T>+collection,+Action<T>+action,+int+degreeOfParallelism)
++++{
++++++++var+blockingCollection+=+new+BlockingCollection<T>();
++++++++foreach+(var+item+in+collection)
++++++++++++blockingCollection.Add(item);
++++++++blockingCollection.CompleteAdding();

++++++++var+tasks+=+new+Task[degreeOfParallelism];
++++++++for+(int+i+=+0;+i+<+degreeOfParallelism;+i%2B%2B)
++++++++{
++++++++++++tasks[i]+=+Task.Factory.StartNew(
++++++++++++++++()+=>
++++++++++++++++{
++++++++++++++++++++foreach+(var+item+in+blockingCollection.GetConsumingEnumerable())
++++++++++++++++++++++++action(item);
++++++++++++++++});
++++++++}
++++++++Task.WaitAll(tasks);
++++}
}|code-block|syntax|javascript|4252979|entityMap|0|LINK|mutability|MUTABLE|url|http://msdn.microsoft.com/en-us/library/dd997411|1|http://msdn.microsoft.com/en-us/library/dd267312.aspx^0|U|H|2D|8|0|11|I|H|I|0|0|D|I|1Y|I|1Y|I|1|0|0^^$0|@$1|2|3|4|5|6|7|Y|8|@$9|Z|A|10|B|C]|$9|11|A|12|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|13|8|@$9|14|A|15|B|C]]|D|@$9|16|A|17|1|18]]|E|$]]|$1|H|3|I|5|6|7|19|8|@$9|1A|A|1B|B|C]|$9|1C|A|1D|B|C]]|D|@$9|1E|A|1F|1|1G]]|E|$]]|$1|J|3|K|5|L|7|1H|8|@]|D|@]|E|$M|N]]|$1|O|3|-4|5|6|7|1I|8|@]|D|@]|E|$]]]|P|$Q|$5|R|S|T|E|$U|V]]|W|$5|R|S|T|E|$U|X]]]]

It's unclear to me what exactly are you doing where “elements are chosen randomly”. But if you use <code>Paralle.ForEach()</code>, then it tries to be efficient and so it partitions the input sequence in some way. If the input sequence is an <code>IList&lt;T&gt;</code>, it will use range partitioning, otherwise, it will use chunk partitioning (see <a href="http://blogs.msdn.com/b/pfxteam/archive/2007/12/02/6558579.aspx" rel="nofollow">Chunk partitioning vs range partitioning in PLINQ</a>).

If you want to process the items in order, you could configure <code>Parallel.ForEach()</code> by using a <a href="http://msdn.microsoft.com/en-us/library/dd997411" rel="nofollow">custom partitioner</a>, which would partition the collection into chunks of size 1.

But since you don't really need <code>Parallel.ForEach()</code> here, possibly a simpler solution would be just to create 4 tasks that process the items one by one. For synchronization, you could use <a href="http://msdn.microsoft.com/en-us/library/dd267312.aspx" rel="nofollow"><code>BlockingCollection</code></a>. Something like:

<pre><code>public static class ParallelOrdered
{
 public static void ForEach&lt;T&gt;(IEnumerable&lt;T&gt; collection, Action&lt;T&gt; action, int degreeOfParallelism)
 {
 var blockingCollection = new BlockingCollection&lt;T&gt;();
 foreach (var item in collection)
 blockingCollection.Add(item);
 blockingCollection.CompleteAdding();

 var tasks = new Task[degreeOfParallelism];
 for (int i = 0; i &lt; degreeOfParallelism; i++)
 {
 tasks[i] = Task.Factory.StartNew(
 () =&gt;
 {
 foreach (var item in blockingCollection.GetConsumingEnumerable())
 action(item);
 });
 }
 Task.WaitAll(tasks);
 }
}
</code></pre>

blocks|key|4342848|text|这就是我完成这项任务的方式|type|unstyled|depth|inlineStyleRanges|entityRanges|data|4342849|public+delegate+void+ProcessFinished(IParallelProcess+process);
public+interface+IParallelProcess
{
++++void+Start();
++++event+ProcessFinished+ProcessFinished;
}

public+class+ParallelProcessBasket+:+ConcurrentQueue<IParallelProcess>
{
++++public+void+Put(IParallelProcess+process)
++++{
++++++++base.Enqueue(process);
++++}
++++public+IParallelProcess+Get()
++++{
++++++++IParallelProcess+process+=+null;
++++++++base.TryDequeue(out+process);
++++++++return+process;
++++}
}
public+class+ParallelProcessor<T>+where+T+:+class
{
++++private+ParallelProcessBasket+basket;
++++private+readonly+int+MAX_DEGREE_OF_PARALLELISM;
++++private+Action<T>+action;
++++public+ParallelProcessor(int+degreeOfParallelism,+IEnumerable<IParallelProcess>+processes,+Action<T>+action)
++++{
++++++++basket+=+new+ParallelProcessBasket();
++++++++this.action+=+action;
++++++++processes.ToList().ForEach(
++++++++++++(p)+=>
++++++++++++{
++++++++++++++++basket.Enqueue(p);
++++++++++++++++p.ProcessFinished+%2B=+new+ProcessFinished(p_ProcessFinished);
++++++++++++});
++++++++MAX_DEGREE_OF_PARALLELISM+=+degreeOfParallelism;
++++}

++++private+void+p_ProcessFinished(IParallelProcess+process)
++++{
++++++++if+(!basket.IsEmpty)
++++++++{
++++++++++++T+element+=+basket.Get()+as+T;
++++++++++++if+(element+!=+null)
++++++++++++{
++++++++++++++++Task.Factory.StartNew(()+=>+action(element));
++++++++++++}
++++++++}
++++}


++++public+void+StartProcessing()
++++{
++++++++//+take+first+level+of+iteration
++++++++for+(int+cnt+=+0;+cnt+<+MAX_DEGREE_OF_PARALLELISM;+cnt%2B%2B)
++++++++{
++++++++++++if+(!basket.IsEmpty)
++++++++++++{
++++++++++++++++T+element+=+basket.Get()+as+T;
++++++++++++++++if+(element+!=+null)
++++++++++++++++{
++++++++++++++++++++Task.Factory.StartNew(()+=>+action(element));
++++++++++++++++}
++++++++++++}
++++++++}
++++}
}
static+void+Main(string[]+args)++++
{
+++++ParallelProcessor<ParallelTask>+pr+=+new+ParallelProcessor<ParallelTask>(Environment.ProcessorCount,+collection,+(e)+=>+e.Method1());
++++++++++++pr.StartProcessing();
}|code-block|syntax|javascript|4342850|谢谢..|4342851|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|L|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|M|8|@]|9|@]|A|$]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

This is how I acheived this task

<pre><code>public delegate void ProcessFinished(IParallelProcess process);
public interface IParallelProcess
{
 void Start();
 event ProcessFinished ProcessFinished;
}

public class ParallelProcessBasket : ConcurrentQueue&lt;IParallelProcess&gt;
{
 public void Put(IParallelProcess process)
 {
 base.Enqueue(process);
 }
 public IParallelProcess Get()
 {
 IParallelProcess process = null;
 base.TryDequeue(out process);
 return process;
 }
}
public class ParallelProcessor&lt;T&gt; where T : class
{
 private ParallelProcessBasket basket;
 private readonly int MAX_DEGREE_OF_PARALLELISM;
 private Action&lt;T&gt; action;
 public ParallelProcessor(int degreeOfParallelism, IEnumerable&lt;IParallelProcess&gt; processes, Action&lt;T&gt; action)
 {
 basket = new ParallelProcessBasket();
 this.action = action;
 processes.ToList().ForEach(
 (p) =&gt;
 {
 basket.Enqueue(p);
 p.ProcessFinished += new ProcessFinished(p_ProcessFinished);
 });
 MAX_DEGREE_OF_PARALLELISM = degreeOfParallelism;
 }

 private void p_ProcessFinished(IParallelProcess process)
 {
 if (!basket.IsEmpty)
 {
 T element = basket.Get() as T;
 if (element != null)
 {
 Task.Factory.StartNew(() =&gt; action(element));
 }
 }
 }


 public void StartProcessing()
 {
 // take first level of iteration
 for (int cnt = 0; cnt &lt; MAX_DEGREE_OF_PARALLELISM; cnt++)
 {
 if (!basket.IsEmpty)
 {
 T element = basket.Get() as T;
 if (element != null)
 {
 Task.Factory.StartNew(() =&gt; action(element));
 }
 }
 }
 }
}
static void Main(string[] args) 
{
 ParallelProcessor&lt;ParallelTask&gt; pr = new ParallelProcessor&lt;ParallelTask&gt;(Environment.ProcessorCount, collection, (e) =&gt; e.Method1());
 pr.StartProcessing();
}
</code></pre>

Thanks..

I have a collection which has elements to be processed and at max only four elements can be processed together. At run time all process started together and all goes in waiting state. Only four elements are processed at one time. 

Problem is the processing elements are chosen randomly as all threads are waiting for resources to get free. Means first element can be the last one from collection.

But, I need to process elements in order they are in the collection.

Please tell me how can I achieve this?

I am using TPL and C# 4.0

How to process items in collection parallely and serially both

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

我有一个集合，它有要处理的元素，最多只有四个元素可以一起处理。在运行时，所有进程一起启动，所有进程都进入等待状态。一次只处理四个元素。问题是处理元素是随机选择的，因为所有线程都在等待资源释放。表示第一个元素可以是集合中的最后一个元素。但是，我需要按照元素在集合中的顺序来处理元素。请告诉我怎样才能做到这一点？我使用的是第...

问如何并行和串行处理集合中的项
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何并行和串行处理集合中的项EN