TPL 控制流问题

Question

我有三个班级：

BargeGrouper 类实现了IBargeGrouper 并具有Group(IEnumerable<Barge> ungrouped) 方法。

BargeGroupMap 类实现了IBargeGroupShow 并具有Show(IEnumerable<BargeGroup> 方法。

如果这两个都是我调用的类：GroupingModule，它是Run()。

问题是，当我调用Group(ungrouped); 然后想Show(bargeGroups); 我在Show(bargeGroups); 中得到IndexOutOfRangeexception，因为IEnumerable 我作为参数传递给它的Count 属性比实际多得多其中的元素。我发现尽管我使用ContinueWith 用BargeGroup 元素填充该集合，但大多数时候分组集合从Group 返回为空。

BargeGrouper的Group方法

public IEnumerable<BargeGroup> Group(IEnumerable<Barge> ungroupedBarges)
{
    List<BargeGroup> bargeGroups = new List<BargeGroup>();
    Int32 groupNumber = 0;

    var riverBarges = from barge in ungroupedBarges
                      where barge != null && !String.IsNullOrEmpty(String.Intern(barge.River))
                      let river = String.Intern(barge.River)
                      orderby river, barge.MileMarker ascending
                      group barge by river into barges
                      select barges.AsEnumerable();

    foreach (IEnumerable<Barge> barges in riverBarges)
    {
        Task.Run(() => ResolveRiver(barges, ref groupNumber)).ContinueWith(t=>
        {
            IEnumerable<BargeGroup> riverGroups = t.Result;
            bargeGroups.AddRange(riverGroups);
        });
    }

    return bargeGroups;
}

ShowBargeGroups:

public void ShowBargeGroups(IEnumerable<BargeGroup> bargeGroups)
{
    Console.WriteLine("Barge groups :");
    if (bargeGroups != null)
    {
        foreach (var group in bargeGroups.Where(b => b != null))
        {
            var title = String.Format("{0}\n\t | {1} \t\t | {2} \t | {3} \t|", group.Id, "Id", "River", "MileMarker");
            Console.WriteLine(title);
            foreach (var barge in group.Barges)
            {
                var caption = String.Format("\t | {0}\t | {1} \t | {2} \t|", barge.Id, barge.River, barge.MileMarker);
                Console.WriteLine(caption);
            }
        }
    }
}

以及GroupingModule:.中的用法。

var groupBarges = bargeGrouper.Group(barges);

bargeGroupShow.ShowBargeGroups(groupBarges);

我能做些什么来解决这个问题？

Answer 1

当使用Task.Run，它返回一个Task而不异步等待它，下一行将立即执行，它不会等待传递的委托完成。

解决此问题的一种方法是并行发出对ResolveRiver 的调用并异步等待多个任务完成：

public Task<IEnumerable<BargeGroup>> GroupAsync(IEnumerable<Barge> ungroupedBarges)
{
    // Do the grouping

    var riverTasks = riverBarges.Select(barges => 
                                        Task.Run(ResolveRiver(barges, ref groupNumber)));

    var result = await Task.WhenAll(riverTasks);
    bargeGroups.AddRange(result.Result.SelectMany(x => x));
    return bargeGroups;
}

然后像这样消费它：

public async Task FooAsync()
{
    var barges = await GroupAsync(ungroupedBarges);
    ShowBargeGroups(barges);
}

注意我会小心ref 参数被并行传递和调用，这不安全。如果您可以从等式中删除groupNumber，请这样做。

Answer 2

Task.Run 不会立即完成它的工作。所以至少你的代码中有一个竞争条件。试试这样：

public async Task<IEnumerable<BargeGroup>> Group(IEnumerable<Barge> ungroupedBarges)
{
    List<BargeGroup> bargeGroups = new List<BargeGroup>();
    Int32 groupNumber = 0;

    var riverBarges = from barge in ungroupedBarges
                      where barge != null && !String.IsNullOrEmpty(String.Intern(barge.River))
                      let river = String.Intern(barge.River)
                      orderby river, barge.MileMarker ascending
                      group barge by river into barges
                      select barges.AsEnumerable();

    foreach (IEnumerable<Barge> barges in riverBarges)
    {
        var riverGroups = await Task.Run(() => ResolveRiver(barges, ref groupNumber));
        bargeGroups.AddRange(riverGroups);
    }

    return bargeGroups;
}

Answer 3

无法保证当您从 Group 方法返回时，Task 已完成对 bargeGroups 列表的处理。我怀疑您的ResolveRiver 方法可能需要一些时间。

在从Group 返回枚举之前，您需要等待执行处理的任务结束。我建议为Task 制作Group async 和awaiting：

public async Task<IEnumerable<BargeGroup>> Group(IEnumerable<Barge> ungroupedBarges)
{
    List<BargeGroup> bargeGroups = new List<BargeGroup>();
    Int32 groupNumber = 0;

    var riverBarges = from barge in ungroupedBarges
                      where barge != null && !String.IsNullOrEmpty(String.Intern(barge.River))
                      let river = String.Intern(barge.River)
                      orderby river, barge.MileMarker ascending
                      group barge by river into barges
                      select barges.AsEnumerable();

    foreach (IEnumerable<Barge> barges in riverBarges)
    {
        await Task.Run(() => ResolveRiver(barges, ref groupNumber)).ContinueWith(t=>
        {
            IEnumerable<BargeGroup> riverGroups = t.Result;
            bargeGroups.AddRange(riverGroups);
        });
    }

    return bargeGroups;
}

确实，正确的方法是使ResolveRiver 异步并等待它。然后你可以去掉你用来与Task通信的bargeGroupsList。

如果您不想采用 async/await 方式，您可以只使用 .Wait() Task。

Answer 4

Task.Run 在你的代码中创建另一个等待你的函数，它很可能会在你的函数返回后被执行。最简单的解决方案是使用 Wait 函数等待它。

Task.Run(() => ResolveRiver(barges, ref groupNumber)).ContinueWith(t=>
    {
        IEnumerable<BargeGroup> riverGroups = t.Result;
        bargeGroups.AddRange(riverGroups);
    }).Wait()

或者事件更好，你不需要Task.Run，​​只等待延续函数就足够了：

ResolveRiver(barges, ref groupNumber)).ContinueWith(t=>
    {
        IEnumerable<BargeGroup> riverGroups = t.Result;
        bargeGroups.AddRange(riverGroups);
    }).Wait();