使用 async/await 等待数据到达数组

Question

我想在现代 nodeJS 工具中有两个并发循环。

一个循环扫描有趣的东西并将它们推送到一个数组中。

第二个循环将数组移出并处理它们。

这个想法是两个循环不会相互阻塞。当第一个循环进行大量扫描但没有找到太多时，第二个循环将能够赶上处理积压。只有当没有积压且没有新数据进来时，处理循环才需要真正等待。

我以多种方式实现了扫描循环。这是容易的部分。 （在这种情况下，我递归地扫描 fs 以查找特定类型的文件。）

对于处理循环，我通过轮询让它工作，但我觉得我应该能够让它与纯异步/等待一起工作。

但我无法理解它。从概念上讲，扫描循环应该履行一个承诺，提醒第二个循环在数组中有东西，或者应该是生成每个新值而不是使用数组的生成器。

但是我看不出如何一遍又一遍地做一个承诺，或者直接等待数组活动，或者作为一个生成器来做而不导致两个循环之间的阻塞。

我一定是想多了！我错过了什么？

---

通过轮询工作的代码，带有可能属于异步/等待实现的注释位：

"use strict";

const { basename, join } = require('path')

const { promisify } = require('util')
const fs = require('fs')
const readdir = promisify(fs.readdir)
const lstat = promisify(fs.lstat)

async function* scanpaths(paths) {
    for (const path of paths) {
        yield* scanonepath(path)
    }
}

async function* scanonepath(path) {
    try {
        const s = await lstat(path)

        if (s.isDirectory()) {
            for (const entry of await readdir(path)) {
                yield* scanonepath(join(path, entry))
            }
        } else if (/\.[mM][pP]3$/.test(path)) {
            yield { pathname:path, basename:basename(path), stat:s }
        }
    } catch (e) {
        // special file, deleted file, etc
    }
}

async function* checkqueue(buf) {
    if (buf.length) {
        yield buf.shift()
    } else {
        // TODO await something to arrive in the buf - HOW?
    }
}

async function processmp3(fullname, name, stat) {
    try {
        console.log(fullname)
        // TODO tricky processing goes here
    } catch (e) {
        console.log(name, e)
    }
}

(async () => {
    let int = null
    let globaldone = false
    let globalprocessing = []
    let buf = []

    async function poll() {
        if (buf.length) {
            let clone = Array.from(buf)
            buf.length = 0  // NOT buf = [] as that doesn't change other refs to buf

            for (let e of clone) {
                globalprocessing.push(processmp3(e.pathname, e.basename, e.stat))
            }
        }

        if (globaldone) {
            await Promise.all(globalprocessing)
            console.warn("*** finished processing")
        } else {
            setTimeout(poll, 125)
        }
    }

    // start polling for scanned files ready to process
    console.log("** start polling")
    setTimeout(poll, 0)

    //console.log("** start scanning")
    //await enqueue(buf)

    globaldone = true
    console.warn("*** finished scanning")
})()

// TODO how?
async function enqueue(buf) {
    // start scanning by iterating over our generator which does the recursive directory stuff
    for await (const file of scanpaths(process.argv.slice(2))) {
        buf.push(file)
        // TODO resolve a promise to notify dequeue? or also yield this file
    }
}

Answer 1

让扫描器循环，当它找到并推送它时，调用一个函数，告诉处理队列某些东西已添加到数组中。如果在处理队列正在做某事时调用该函数，则什么也不做 - 否则，如果处理队列空闲，这将告诉处理队列从扫描数组中转移一个项目并处理它。

当然，当处理队列处理完一个项目后，让它检查是否有额外的项目需要处理，如果有，立即处理下一个。这样可以确保处理队列仅在没有更多要处理的项目时才会空闲，并且一旦推送了新项目，它将再次开始运行。

例如，使用您的代码，您可以改为执行以下操作：

let processing = false;
async function processNext() {
  if (processing) {
    return;
  }
  processing = true;
  const e = buf.shift();
  const mp3 = await processmp3(e.pathname, e.basename, e.stat);
  // do something with parsed mp3?
  processing = false;
  processNext();
}


async function enqueue(buf) {
  for await (const file of scanpaths(process.argv.slice(2))) {
    buf.push(file);
    processNext();
  }
}