等效于 do() while{} 并行

Question

如何在下面的 Update() 方法中创建 do-while 或类似的并行等效项？

应用程序中的另一个线程随机写入TestBuffer。 TestBuffer.RemoveItemAndDoSomethingWithIt(); 应该一直运行到 TestBuffer 为空。目前Update() 仅与枚举时集合中的项目一起运行，这是有道理的。

internal class UnOrderedBuffer<T> where T : class
{
    ConcurrentBag<T> GenericBag = new ConcurrentBag<T>();
}

internal class Tester
{
    private UnOrderedBuffer<Data> TestBuffer;

    public void Update()
    {
        Parallel.ForEach(TestBuffer, Item =>
        {
            TestBuffer.RemoveItemAndDoSomethingWithIt();
        });
    }
}

Answer 1

您可以通过“添加”一个空值/默认值来强制执行一次：

static IEnumerable<T> YieldOneDefault<T>(this IEnumerable<T> values)
{
    yield return default(T);
    foreach(var item in values)
        yield return item;
}

然后按如下方式使用：

Parallel.ForEach(TestBuffer.YieldOneDefault(), Item =>  
{  
    if(Item != null)
      TestBuffer.RemoveItemAndDoSomethingWithIt();
    else
      DoSomethingDuringTheFirstPass();
});  

虽然我怀疑您可能正在寻找以下扩展方法：

public static IEnumerable<IEnumerable<T>> GetParrallelConsumingEnumerable<T>(this IProducerConsumerCollection<T> collection)
{
    T item;
    while (collection.TryTake(out item))
    {
        yield return GetParrallelConsumingEnumerableInner(collection, item);
    }
}

private static IEnumerable<T> GetParrallelConsumingEnumerableInner<T>(IProducerConsumerCollection<T> collection, T item)
{
    yield return item;
    while (collection.TryTake(out item))
    {
        yield return item;
    }
}

这会让你得到这个结果（我认为这就是你所追求的）：

Parallel.ForEach(TestBuffer.GetParrallelConsumingEnumerable(), Items =>       
{
    foreach(var item in Items)
    {
       DoSomethingWithItem(item);
    }
});

Answer 2

for/foreach 通常用于对多个项目执行任务。

while-do/do-while 用于：

一个。对多个项目执行任务 尚未枚举的（例如一棵树）。
- 在这种情况下，您可以定义 BFS 或 DFS 枚举器并在 foreach 中使用。

b.对单个项目执行迭代工作
- 迭代工作不适合并行性

不要尝试将代码从串行重构为并行。相反，请考虑您的任务是什么以及如何最好地并行完成。 （重构算法，而不是代码。）

Answer 3

public static void While( 
    ParallelOptions parallelOptions, Func<bool> condition, 
    Action<ParallelLoopState> body) 
{ 
    Parallel.ForEach(Infinite(), parallelOptions, (ignored, loopState) => 
    { 
        if (condition()) body(loopState); 
        else loopState.Stop(); 
    }); 
}

private static IEnumerable<bool> Infinite() 
{ 
    while (true) yield return true; 
}