Как заблокировать веб-сокет (или состояние), пока скребок не будет сделан

Question

У меня есть несколько скребков, как показано ниже:

await scraper1.run
await scraper2.run
// etc

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

const express = require('express')
const app = express()
const http = require('http').Server(app)
const cors = require('cors')
const puppeteer = require('puppeteer')
const io = require('socket.io')(http)
const mongoose = require('mongoose')
const _ = require('lodash')

const scraper1 = require('./scraper1')
const scraper2 = require('./scraper2')

mongoose.connect("mongodb://localhost:27017/test")

;(async function () {
  try {
    const browser = await puppeteer.launch({
      headless: false
    })

    io.on('connection', async function (socket) {
      socket.on('search', async function (query) {
        // check whether document exists with user ip address then return
        // otherwise run the scrapres
        await scraper1.run(browser, socket, query)
        await scraper2.run(browser, socket, query)
      })
    })
  } catch (e) {
    console.log(e)
  }
})()

http.listen(3000)

Контекст : когда пользователь выполняет обновление и несколько соединений с сокетами, скребки запускаются несколько раз и данные дублируются.Я предотвратил дублирование с помощью mongodb, но проблема с производительностью остается, потому что скребки работают до тех пор, пока их результаты не будут готовы, затем я проверяю базу данных.

Вопрос : Как заблокировать или запретить запуск скребков несколько раза также подождать, пока каждый скребок будет сделан с websocket?

Answer 1

Я могу предложить вам следующее решение (я не проверял его, но, думаю, вы можете понять, чего я пытаюсь достичь, объяснение ниже примера):

'use strict';

const queryState = {

};

const getQueryKey = (query) => {
  // base64 but can be a hash like sha256
  const key = Buffer.from(query).toString('base64');

  return key;
};

/**
 * Return query state
 * @param {String} query
 * @return {String} state [PENDING, DONE, null] null if query doesn't exist
 */
const getQueryState = (query) => {
  const key = getQueryKey(query);

  const state = queryState[key] || null;

  return state;
};

/**
 * Add a query and initialize it as pending
 * @param {String} query
 * @return {String} state
 */
const addQuery = (query) => {
  const key = getQueryKey(query);
  const state = 'PENDING';
  queryState[key] = state;

  return state;
};

/**
 * Hashmap to associate pending queries to be notified to socket connections
 * when query is done
 * This structure keeps and array of callbacks per query key
 */
const observers = {

};

const addObserver = (query, callback) => {
  const key = getQueryKey(query);

  if (typeof observers[key] !== 'undefined') {
    observers[key] = [callback];
  } else {
    observers[key] = [...observers[key], callback];
  }
};

const notifyObservers = (query) => {
  const key = getQueryKey(query);

  const callbacks = observers[key] || [];

  // TODO: get query data scrapper from a cache / database / etc
  const data = getDataFromQuery(query);
  callbacks.forEach((callback) => {
    callback(data);
  });
};

/**
 * Update query status to done
 * PreCD: query must exist in queryState (previously added using addQuery)
 * @param {String} query
 * @return {String} state
 */
const endQuery = (query) => {
  const key = getQueryKey(query);
  const state = 'DONE';
  queryState[key] = state;

  return state;
};

io.on('connection', async function (socket) {
  socket.on('search', async function (query) {
    /**
     * If query doesn't exist, scrap it
     */
    const state = getQueryState(query);
    if (state === null) {
      addQuery(query);
      await scraper1.run(browser, socket, query);
      await scraper2.run(browser, socket, query);
      endQuery(query);

      // store scrapper data in cache / database / etc and
      // socket send scraperData to the user

      // notify pending queries to send data scrapper
      notifyObservers(query);
    } else if (state === 'PENDING') {
      // add callback to return data to the user
      addObserver(query, (scraperData) => {
        // socket send scraperData to the user
      });
    } else {
      // socket send scraperData to the user
    }
  });
});

Чтобы упростить вещиПример простой, но не самый лучший с точки зрения производительности / архитектуры.Это решение реализует:

Сценарий 1 (кто-то впервые запрашивает запрос1)

1. В бэкэнд поступил запрос (соединение с сокетом) с запросом
2. Thisзапрос еще не существует, поэтому отметьте его как PENDING и запустите scrappers

Сценарий 2 (еще один запрашивает запрос1)

1. Появляется второе соединение с запросом того жезапрос как сценарий 1
2. Запрос находится в состоянии PENDING, поэтому мы добавляем обратный вызов для вызова, когда этот запрос завершится

Сценарий 3 (запрос 1 завершен)

1. Сценарии начаты в сценарии 1 и заканчиваются, поэтому запрос1 продается как DONE
2. Каждый запрос (наблюдатель), ожидающий запрос1, будет уведомлен

Это решение может иметь несколько способов:быть реализованным, но моя точка зрения состоит в том, чтобы попытаться раскрыть это вам, а затем вы можете изменить его так, как хотите.

Надеюсь, это поможет