设计协程线程实现的替代方案，将推送方法转换为拉取方法

Question

我有一个集合类，它以压缩格式保存许多不同类型的数据。为了枚举集合中的所有值，有一个 Execute(Query, IDataWriter) 方法。您向它传递一个定义所需数据的查询，然后对于每条匹配数据，它都会调用您传入的 IDataWriter 对象上的一个方法。IDataWriter

接口有 15 种不同的方法，每个方法对应集合中的每种不同数据类型。现在我需要将此数据提交到数据库，并且我希望能够实现 IEnumerator将数据提交到数据库。问题在于如何将调用将大量数据转储到 IDataWriter 对象（push）的 Execute 转换为 pull 方法，以便可以使用 IEnumerator 的 MoveNext 和 Current。

我研究过协程和纤程，但我发现的所有示例似乎都不适用于内部对协程一无所知的现有方法（在我的例子中为 Execute）。所以我的伪代码计划是使用线程和手动同步执行以下操作。

Execute 方法将在单独的线程中运行，我将在每个 IDataWriter 方法中手动等待并发出信号，

class EnumeratorAdapterObject : IEnumerator<SqlDataRecord>, IDataWriter
{

public EnumeratorAdapterObject(Store storeObject)
{
    workerThread = new Thread(storeObject.Execute(query, this));
}

public bool MoveNext()
{
    if (firstTimeCalled)
    {
        start worker thread
    }
    else
    {
        signal resume
    }
    block for either a call into an Add method or the Execute thread finishes

    if (not anything buffered)
        return false
    else
        return true
}


// 14 other methods like this implemented in IDataWriter, each with different types
public void Add_Decimal(IntvlDataHeader header, decimal data)
{
    buffer field of current SqlDataRecord = generate record;

    signal main thread
    wait for resume signal
}

public SqlDataRecord Current
{
    get { return buffer field of current SqlDataRecord; }
}

}

这看起来是一个好方法吗？有谁知道已经实现了这一点的任何示例或问题？

或者是否有办法利用 4.0 的新功能？我考虑过使用阻塞并发集合，其中限制为 1 个东西，但是消费者（IEnumerator 的 MoveNext）如何知道另一个线程何时完成添加内容？

Answer 1

我发现我可以使用阻塞集合，并在完成后调用 CompleteAdding() ，而不是使用 Signal/Wait 进行手动线程创建和同步。下面是一个简单的示例，对于我上面的问题，我将其包装在一个实现 IEnumerator和 IDataWriter 的对象中，因此我将使用调用 Execute 和 Add_* 方法将调用 col.Add(new SqlDataRecord(....))

static void Main(string[] args)
{
    var col = new BlockingCollection<int>(1);

    Task.Factory.StartNew(
        () =>
        {
            GenerateStuff(col);
            col.CompleteAdding();
        });

    while (!col.IsCompleted)
    {
        Thread.Sleep(100);

        int result;
        if (!col.TryTake(out result, -1))
        {
            break;
        }
        Console.WriteLine("Got {0}", result);
    }

    Console.WriteLine("Done Adding!");
}

static void GenerateStuff(BlockingCollection<int> col)
{
    for (int i = 0; i < 10; i++)
    {
        Thread.Sleep(10);
        Console.WriteLine("Adding {0}", i);
        col.Add(i);
        Console.WriteLine("Added {0}", i);
    }
}

这还有一个优点，即运行 Execute 的工作线程将生成下一个结果与 IEnumerator 返回 Current 和 sql 代码同时执行其提交数据的操作。通过手动线程信号，一次只能运行一个线程。