3

我 n在我的函数中打开并发线程：

List<string> _files = new List<string>();

public void Start()
{
    CancellationTokenSource _tokenSource = new CancellationTokenSource();
    var token = _tokenSource.Token;

    Task.Factory.StartNew(() =>
    {
        try
        {
            Parallel.ForEach(_files,
                new ParallelOptions
                {
                    MaxDegreeOfParallelism = 5 //limit number of parallel threads 
                },
                file =>
                {
                    if (token.IsCancellationRequested)
                        return;
                    //do work...
                });
        }
        catch (Exception)
        { }

    }, _tokenSource.Token).ContinueWith(
        t =>
        {
            //finish...
        }
    , TaskScheduler.FromCurrentSynchronizationContext() //to ContinueWith (update UI) from UI thread
    );
        }

线程打开后，我注意到它从我的列表中选择随机文件。是否可以每次n从我的列表中选择第一个元素？

4

3 回答 3

4

要获得您想要的行为，您需要编写一个自定义分区程序，它看起来“随机”的原因是现在它正在批量处理块中的文件列表，所以如果您的源列表是

List<string> files = List<string> { "a", "b", "c", "d", "e", "f", "g", "h", "i" };

当它对它进行分区时，它可能会像这样均匀地分割它（如果 Max 是 3 个线程）：

  • Thread1的工作清单：“a”、“b”、“c”
  • Thread2的工作清单：“d”、“e”、“f”
  • Thread3的工作清单：“g”、“h”、“i”

因此，如果您查看正在处理的文件，它可能看起来像

"a", "d", "g", "e", "b", "h", "c", "f", "i"

如果您制作自定义分区器，您可以让它一次处理一个项目，而不是一次处理一批，以使工作列表看起来像

  • Thread1的工作清单：“a”，GetTheNextUnprocessedString()
  • Thread2的工作清单：“b”，GetTheNextUnprocessedString()
  • Thread3的工作清单：“c”，GetTheNextUnprocessedString()

如果您使用的是 .NET 4.5，您可以像这样使用这个工厂：

Parallel.ForEach(Partitioner.Create(_files, EnumerablePartitionerOptions.NoBuffering),
                new ParallelOptions
                {
                    MaxDegreeOfParallelism = 5 //limit number of parallel threads 
                },
                (file, loopstate, index) =>
                {
                    if (token.IsCancellationRequested)
                        return;
                    //do work...
                });

如果您不使用 .NET 4.5，这不是一项简单的任务，所以我不会在这里为您编写它。阅读我在顶部链接的 MSDN 文章，您最终将能够弄清楚。

我要做的是问自己“我真的需要按顺序处理文件吗？” 如果您不需要它们按顺序排列，请让它自己订购，因为您通过执行订单可能会做的唯一一件事是可能会减慢流程。

于 2013-10-31T14:45:04.840 回答
3

只是不要依赖Parallel.ForEach以特定顺序启动工作项是否重要；正如其他人所说，您可以根据需要进行配置，但这并不容易。

更简单的选择是只创建 5 个不同的任务来处理这些项目。它没有根据需要动态添加/删除工作人员的能力，但无论如何您似乎都没有充分利用这一点。

只需创建一个BlockingCollection和 5 个从中获取项目的任务：

var queue = new BlockingCollection<string>();
int workers = 5;
CancellationTokenSource cts = new CancellationTokenSource();
var tasks = new List<Task>();

for (int i = 0; i < workers; i++)
{
    tasks.Add(Task.Run(() =>
    {
        foreach (var item in queue.GetConsumingEnumerable())
        {
            cts.Token.ThrowIfCancellationRequested();

            DoWork(item);
        }
    }, cts.Token));
}

//throw this into a new task if adding the items will take too long
foreach (var item in data)
    queue.Add(item);
queue.CompleteAdding();

Task.WhenAll(tasks).ContinueWith(t =>
{
    //do completion stuff
});
于 2013-10-31T15:00:50.063 回答
0

当然文件是随机选择的，这就是parallel.foreach 的重点。如果您并行，您指定的 5 个线程将使用输入，因为它由数据分区器决定。

但是，如果您真的想保持顺序，请检查您可以为 parallel.foreach 指定的 OrderablePartitioner。-> http://msdn.microsoft.com/en-us/library/dd989583.aspx 当然这会降低性能，但它允许您指定如何为线程创建分区。

于 2013-10-31T14:39:42.910 回答