如何在 JS/TS 中实现伪阻塞异步队列? [英] How to implement a pseudo blocking async queue in JS/TS?

问题描述

所以这里有一个矛盾:我想在 javascript/typescript 中创建一个异步阻塞队列(如果你可以在没有 typescript 的情况下实现它，那很好).基本上我想实现一些类似于 Java 的 BlockingQueue 期望而不是它实际上被阻塞，它将是异步的，我可以等待出队.

So here's an oxymoron: I want to create an asynchronous blocking queue in javascript/typescript (if you can implement it without typescript, that's fine). Basically I want to implement something like Java's BlockingQueue expect instead of it actually being blocking, it would be async and I can await dequeues.

这是我要实现的接口:

interface AsyncBlockingQueue<T> {
  enqueue(t: T): void;
  dequeue(): Promise<T>;
}

我会像这样使用它:

// enqueue stuff somewhere else

async function useBlockingQueue() {
  // as soon as something is enqueued, the promise will be resolved:
  const value = await asyncBlockingQueue.dequeue();
  // this will cause it to await for a second value
  const secondValue = await asyncBlockingQueue.dequeue();
}

有什么想法吗?

推荐答案

其实很简单，dequeue 会创建一个promise，enqueue 会解决这个问题.我们只需要将解析器保持在队列中 - 并且还关心值在出队之前入队的情况，将已经履行的承诺保持在队列中.

It's quite simple actually, dequeue will create a promise that enqueue will resolve. We just have to keep the resolvers in a queue - and also care about the case where values are enqueued before they are dequeued, keeping the already fulfilled promises in a queue.

class AsyncBlockingQueue {
  constructor() {
    // invariant: at least one of the arrays is empty
    this.resolvers = [];
    this.promises = [];
  }
  _add() {
    this.promises.push(new Promise(resolve => {
      this.resolvers.push(resolve);
    }));
  }
  enqueue(t) {
    // if (this.resolvers.length) this.resolvers.shift()(t);
    // else this.promises.push(Promise.resolve(t));
    if (!this.resolvers.length) this._add();
    this.resolvers.shift()(t);
  }
  dequeue() {
    if (!this.promises.length) this._add();
    return this.promises.shift();
  }
  // now some utilities:
  isEmpty() { // there are no values available
    return !this.promises.length; // this.length <= 0
  }
  isBlocked() { // it's waiting for values
    return !!this.resolvers.length; // this.length < 0
  }
  get length() {
    return this.promises.length - this.resolvers.length;
  }
  [Symbol.asyncIterator]() {
    // Todo: Use AsyncIterator.from()
    return {
      next: () => this.dequeue().then(value => ({done: false, value})),
      [Symbol.asyncIterator]() { return this; },
    };
  }
}

我不知道 TypeScript，但大概添加必要的类型注释很简单.

I don't know TypeScript, but presumably it's simple to add the the necessary type annotations.

为了获得更好的性能，请使用带有循环缓冲区而不是普通数组的队列实现，例如这个.您也可以只使用一个队列并记住您当前是存储承诺还是解析器.

For better performance, use a Queue implementation with circular buffers instead of plain arrays, e.g. this one. You might also use only a single queue and remember whether you currently store promises or resolvers.

这篇关于如何在 JS/TS 中实现伪阻塞异步队列?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！