在两个线程上交错进程

本文关键字：进程 线程 两个 | 更新日期: 2023-09-27 17:56:41

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

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

有没有办法在第二个线程上运行查询/回调并将该代码与迭代器代码交错？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中可用的大多数锁来实现这一点，但它们都不起作用。我记得尝试过信号量，信号量Slim，AutoResetEvent，ManualResetEvent和Mutex。

PS：DoQueryWithCallback 来自库。它是一个本机库（ILSpy/Reflector/etc无法反编译它）。我想函数看起来像这样：

long DoQueryWithCallback(string queryString, Callback rowCallback)
{
    do some setup;
    Row row;
    while (next(out row))
            rowCallback(row);
    do some teardown;
}

在两个线程上交错进程

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

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

下面是一个使用 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();
    }
}
相关文章：