c# - 在两个线程上交错处理

Question

我有一个库，可以用来访问一些表格数据。这个库是我访问数据的唯一途径。我使用的方法采用查询字符串和为每个结果行调用的回调。

目前，回调将每一行加载到一个列表中，然后返回该列表。我想使用迭代器模式，但我对数据的唯一访问权限是通过此回调方法。

有没有办法在第二个线程上运行查询/回调并将该代码与迭代器代码交错？ Psudocode:

IEnumerable<Row> QueryData(string queryString)
{
    var callerLock = create new sync lock;
    var callbackLock = create new sync lock;
    var rows = create new stack of rows with capacity 1;
    var qthread = create new thread with QueryCallback(queryString, callerLock, callbackLock, rows);

    start qthread;
    while (qthread is running)
    {
        signal callbackLock;
        wait for callerLock;

        if stack is empty
            break;

        var row = pop from rows;
        yield return row;
    }
}

void QueryCallback(string queryString, lock callerLock, lock callbackLock, Stack<Row> rows)
{
    DoQueryWithCallback(queryString, row =>
    {
        wait for callbackLock;
        push row to rows;
        signal callerLock;
    });

    signal callerLock;
}

我曾尝试使用 .NET Framework 中的大部分可用锁来实现这一点，但它们都不起作用。我记得尝试过 Semaphore、SemaphoreSlim、AutoResetEvent、ManualResetEvent 和 Mutex。

P.S.:DoQueryWithCallback 来自图书馆。它是一个本地库(ILSpy/Reflector/etc 不能反编译它)。我想这个函数看起来像这样:

long DoQueryWithCallback(string queryString, Callback rowCallback)
{
    do some setup;

    Row row;
    while (next(out row))
            rowCallback(row);

    do some teardown;
}

Answer 1

如果我对伪代码的理解正确，您希望在后台线程上触发提取操作并在结果进入时使用迭代器生成结果，而不是等待整个提取操作完成再返回。我要更改的几件事:

• 如果要保持获取行的顺序，请使用队列而不是堆栈
• 信号/阻塞只需要一种方式——线程产生返回行需要等待获取线程将项目添加到队列中。无需阻塞抓取线程

这是一个使用 Task、ConcurrentQueue 和 AutoResetEvent 的简单示例:

public IEnumerable<Row> GetRows(string query)
{
    using (var resetEvent = new AutoResetEvent(false))
    {
        var rows = new ConcurrentQueue<Row>();

        var queryTask = Task.Run(() => DoQueryWithCallback(query, r =>
        {
            rows.Enqueue(r);
            resetEvent.Set();
        }));
        queryTask.ContinueWith(t => resetEvent.Set()); // This ensures that queryTask.IsCompleted will be true in the while loop below

        while (resetEvent.WaitOne() && !queryTask.IsCompleted)
        {
            Row row;
            while (rows.TryDequeue(out row))
                yield return row;
        }
    }
}

编辑

实际上有一个更好的方法使用 BlockingCollection

public IEnumerable<Row> GetRows(string query)
{
    using (var rows = new BlockingCollection<Row>())
    {
        Task.Run(() =>
        {
            DoQueryWithCallback(query, r => rows.Add(r));
            rows.CompleteAdding();
        });

        while (!rows.IsCompleted)
            yield return rows.Take();
    }
}