Блог

Из вашего описания я понимаю, что следует:

• У вас есть Node.js первый ярус. Каждый Node.js поле должно быть ограничено до 100 клиентами
• У вас есть неопределенный задний уровень, который имеет GRPC-соединения с блоками на переднем уровне (назовем их EAP). Каждый ВП <-> Node.js Связь GRPS ограничена N одновременными подключениями.

То, что я вижу здесь,-это ограничения только на уровне сервера и на уровне подключения, поэтому я не вижу причин иметь какую-либо распределенную систему (например, Bull) для управления очередью (если Node.js поле умирает, никто не может восстановить контекст HTTP — запроса, чтобы предложить ответ на этот конкретный запрос, поэтому, когда Node.js блок умирает, ответы на его запросы не более полезны).

Учитывая это, я бы просто создал локальную очередь (такую же простую, как массив) для управления вашей очередью.

Отказ от ответственности: это следует считать псевдокодом, то, что следует ниже, упрощено и непроверено

Это может быть реализация очереди:

 interface SimpleQueueObject<Req, Res> {
  req: Req;
  then: (Res) => void;
  catch: (any) => void;
}


class SimpleQueue<Req = any, Res = any> {

  constructor(
    protected size: number = 100,
    /** async function to be executed when a request is de-queued */
    protected execute: (req: Req) => Promise<Res>,
    /** an optional function that may ba used to indicate a request is
     not yet ready to be de-queued. In such case nex request will be attempted */
    protected ready?: (req: Req) => boolean,
  ) { }

  _queue: SimpleQueueObject<Req, Res>[] = [];
  _running: number = 0;

  private _dispatch() {
    // Queues all available
    while (this._running < this.size amp;amp; this._queue.length > 0) {
      // Accept
      let obj;
      if (this.ready) {
        const ix = this._queue.findIndex(o => this.ready(o.req));
        // todo : this may cause queue to stall (for now we throw error)
        if (ix === -1) return;
        obj = this._queue.splice(ix, 1)[0];
      } else {
        obj = this._queue.pop();
      }
      // Execute
      this.execute(obj.req)
        // Resolves the main request
        .then(obj.then)
        .catch(obj.catch)
        // Attempts to queue something else after an outcome from EAP
        .finally(() => {
          this._running --;
          this._dispatch();
        });
      obj.running = true;
      this._running   ;
    }
  }

  /** Queue a request, fail if queue is busy */
  queue(req: Req): Promise<Res> {
    if (this._running >= this.size) {
      throw "Queue is busy";
    }

    // Queue up
    return new Promise<Res>((resolve, reject) => {
      this._queue.push({ req, then: resolve, catch: reject });
      this._dispatch();
    });
  }

  /** Queue a request (even if busy), but wait a maximum time
   * for the request to be de-queued */
  queueTimeout(req: Req, maxWait: number): Promise<Res> {
    return new Promise<Res>((resolve, reject) => {
      const obj: SimpleQueueObject<Req, Res> = { req, then: resolve, catch: reject };
      // Expire if not started after maxWait
      const _t = setTimeout(() => {
        const ix = this._queue.indexOf(obj);
        if (ix !== -1) {
          this._queue.splice(ix, 1);
          reject("Request expired");
        }
      }, maxWait);
      // todo : clear timeout
      // Queue up
      this._queue.push(obj);
      this._dispatch();
    })
  }

  isBusy(): boolean {
    return this._running >= this.size;
  }

}
 

А затем ваш Node.js бизнес-логика может делать что-то вроде:

 const EAP1: SimpleQueue = /* ... */;
const EAP2: SimpleQueue = /* ... */;

const INGRESS: SimpleQueue = new SimpleQueue<any, any>(
  100,
  // Forward request to EAP
  async req => {
    if (req.forEap1) {
      // Example 1: this will fail if EAP1 is busy
      return EAP1.queue(req);
    } else if (req.forEap2) {
      // Example 2: this will fail if EAP2 is busy and the request can not
      // be queued within 200ms
      return EAP2.queueTimeout(req, 200);
    }
  }
)

app.get('/', function (req, res) {
  // Forward request to ingress queue
  INGRESS.queue(req)
    .then(r => res.status(200).send(r))
    .catch(e => res.status(400).send(e));
})

 

Или это решение позволит вам (по запросу) также принимать запросы на загруженные EAP (в общей сложности не более 100) и отправлять их, когда они будут готовы:

 const INGRESS: SimpleQueue = new SimpleQueue<any, any>(
  100,
  // Forward request to EAP
  async req => {
    if (req.forEap1) {
      return EAP1.queue(req);
    } else if (req.forEap2) {
      return EAP2.queue(req);
    }
  },
  // Delay queue for busy consumers
  req => {
    if (req.forEap1) {
      return !EAP1.isBusy();
    } else if (req.forEap2) {
      return !EAP2.isBusy();
    } else {
      return true;
    }
  }
)
 

Пожалуйста, обратите внимание, что:

• в этом примере, Node.js начнет выбрасывать, когда будет получено более 100 одновременных запросов (нет ничего необычного в том, чтобы выбросить 503 при регулировании)
• Будьте осторожны, когда у вас больше ограничений на регулирование (Node.js и GRPC в вашем случае), поскольку первое может привести к секундному голоданию (подумайте о получении 100 запросов для EAP1, а затем 10 для EAP2, Node.js будет заполнен запросами EAP1 и откажется от запросов EAP2 все, что делает EAP2, ничего не делает)