c# - 多个线程使用 yield 访问 IEnumerable

Question

我正在使用第三方库迭代一些非常大的平面文件，这可能需要很长时间。该库提供了一个枚举器，因此您可以生成每个结果并对其进行处理，而枚举器随后会提取平面文件中的下一项。

例如:

IEnumerable<object> GetItems()
{
    var cursor = new Cursor;

    try
    {
        cursor.Open();

        while (!cursor.EOF)
        {
            yield return new //object;

            cursor.MoveNext();
        }

    }
    finally
    {
        if (cursor.IsOpen)
        {
            cursor.Close();
        }
    }
}

我想要实现的是让两个消费者使用同一个 Enumerable，这样我就不必两次提取信息，这样每个消费者仍然可以在每个项目到达时处理它，而不必一直等待立即到达。

IEnumerable<object> items = GetItems();

new Thread(SaveToDateBase(items)).Start();
new Thread(SaveSomewhereElse(items)).Start();

我想我想要实现的是类似

“如果消费者要求的项目已经被提取，则放弃它，否则移动下一步并等待”但我意识到两个线程之间可能发生 MoveNext() 冲突。

如果没有任何关于如何实现的想法，这样的事情是否已经存在？

谢谢

Answer 1

Pipelines pattern implementation使用 .NET 4 BlockingCollection<T>而 TPL Tasks 正是您要找的。请参阅我的完整示例答案 in this StackOverflow post .

示例:3 个同时存在的消费者

BlockingCollection<string> queue = new BlockingCollection<string>();    
public void Start()
{
    var producerWorker = Task.Factory.StartNew(() => ProducerImpl());
    var consumer1 = Task.Factory.StartNew(() => ConsumerImpl());
    var consumer2 = Task.Factory.StartNew(() => ConsumerImpl());
    var consumer3 = Task.Factory.StartNew(() => ConsumerImpl());

    Task.WaitAll(producerWorker, consumer1, consumer2, consumer3);
}

private void ProducerImpl()
{
   // 1. Read a raw data from a file
   // 2. Preprocess it
   // 3. Add item to a queue
   queue.Add(item);
}

// ConsumerImpl must be thrad safe 
// to allow launching multiple consumers simulteniously
private void ConsumerImpl()
{
    foreach (var item in queue.GetConsumingEnumerable())
    {
        // TODO
    }
}

如果还有什么不清楚的，请告诉我。

管道流的高级图: