WebSockets - большой поток данных

[NatalliaF]

Клиент и сервер общаются посредством WebSockets. Сервер передает клиенту непрерывно сообщения. Клиент их получает в onmessage. Стандартно. Но! Сообщений о-очень много. Очередное сообщение может быть передано сервером, когда клиент еще не закончил обрабатывать предыдущее. Может, спрошу очевидную вещь. Но вопрос мой такой. Когда в процессе обработки полученного в onmessage сообщения приходит еще одно сообщение, то создается новый поток и код в onmessage для нового сообщения выполняется в этом новом потоке или все происходит последовательно (пока предыдущее сообщение не обработается, новое обрабатываться не будет)?

[voraa]

Конечно последовательно. javascript в принципе не предназначен для параллельной работы. Ведь функции, которые вызываются по onmessage, разделяют (им доступны) все переменные и объекты. Их параллельное изменение может нарушить и механизмы оптимизации и сборку мусора.
Только worker может работать параллельно в другом потоке, но он полностью изолирован, использует другую область видимости.

[NatalliaF]

Т.е. в onmessage при получении очередного сообщения я его просто обрабатываю и уверена, что пока не закончу обработку, сервер больше ничего не отправит. А если хочу обрабатывать быстро и параллельно, то, получая сообщение в onmessage, каким-то образом должна создать что-то типа отдельного потока, передать ему сообщение и быстро выйти из onmessage, чтобы сервер продолжил передавать данные.
Веб-воркеры - потоки, принадлежащие браузеру. А у меня клиент не браузерный. Выходит, надо искать что-то другое.

[voraa]

Т.е. в onmessage при получении очередного сообщения я его просто обрабатываю и уверена, что пока не закончу обработку, сервер больше ничего не отправит. А если хочу обрабатывать быстро и параллельно, то, получая сообщение в onmessage, каким-то образом должна создать что-то типа отдельного потока, передать ему сообщение и быстро выйти из onmessage, чтобы сервер продолжил передавать данные.
Веб-воркеры - потоки, принадлежащие браузеру. А у меня клиент не браузерный. Выходит, надо искать что-то другое.

Нет. Такой уверенности быть не должно. Сервер может отправлять данные когда ему вздумается. Данные будут буферизироваться и обрабатываться клиентом по мере готовности.
Чтобы обеспечить параллельную обработку (в другом потоке) нужен Worker, они есть не только в браузерах. В Node тоже есть.

[SuperZen]

package.json

{
  "name": "ws-seq-para",
  "version": "1.0.0",
  "main": "index.js",
  "license": "MIT",
  "dependencies": {
    "ws": "^7.2.3"
  }
}

server.js

const WebSocket = require('ws')
const wss = new WebSocket.Server({ port: 2999 })
wss.on('connection', function connection(ws) {
  ws.send(JSON.stringify({ type: 'long' }))
  setTimeout(() => {
    ws.send(JSON.stringify({ type: 'short' }))
  }, 1000)
})

client.js

const WebSocket = require('ws')
const ws = new WebSocket('ws://localhost:2999')
ws.on('message', function incoming(json) {
  const { type } = JSON.parse(json)
  switch (type) {
    case 'long':
      console.log('long start')
      setTimeout(console.log, 2000, 'long stop')
      break
    case 'short':
      console.log('short running')
      break
    default:
      break
  }
})

вывод консоли:
long start
short running
long stop

резюме: нужен queue... воркер не решает эту проблему...

[voraa]

резюме: нужен queue... воркер не решает эту проблему...

Если сообщения требуется обрабатывать именно в порядке их поступления, но поступают они чаще, чем их возможно обработать, то и queue не решит эту проблему.
А если их можно обрабатывать независимо, то можно и несколько воркеров запустить и передавать тому, который освободился.

[SuperZen]

Если сообщения требуется обрабатывать именно в порядке их поступления, но поступают они чаще, чем их возможно обработать, то и queue не решит эту проблему.

сообщения поступают, и их надо обрабатывать именно в порядке поступления, то нужен queue

А если их можно обрабатывать независимо, то можно и несколько воркеров запустить и передавать тому, который освободился.

это да

[NatalliaF]

Спасибо, ребята, за информацию и идеи.

Выше упоминалось, что если сервер посылает сообщения быстрее, чем они обрабатываются, то на клиенте сообщения буферизуются. Отсюда возник еще один вопрос. Если клиента надо отключить (он прекратит принимать сообщения от сервера), как при этом извлечь из буфера имеющиеся сообщения с целью их обработки (информацию терять нельзя)?

[voraa]

Очень абстрактный вопрос.
Получение сообщения начинается с сетевой карты и ее драйвера. Потом это сообщение передается ОС, потом ОС передает его клиенту (браузеру, еще чему то...) Потом браузер передает его программе на javascrit в виде события message. И если отключить клиента до этого момента (как? снять задачу? выключить компьютер?) то с этим сообщением мы ничего сделать не сможем. Оно в любом случае потеряется.
А вот с теми, которые пришли в качестве событий message уже можно разбираться.
Как мне представляется у события есть следующие фазы:
Прием (вызывается функция, указанная в onmessage)
Обработка (Какая то длительная операция)
Использование результата обработки (Сравнительно короткая операция)

Ну, например, мы принимаем какие то массивы
Обработка - это сортировка массива
Использование - вывод массива на экран.

Можно написать функцию
onmessage = function (ev) {
sort(0)
output ()
}
Если, скажем sort - занимает 1 сек, а сообщения приходят каждые 300 мс, то у нас выстроится очередь из необработанных событий. Т.к функция обработки события выполняется в основном потоке, и обрабатывать следующее событие мы можем только после того, как закончится обработка текущего.
Допустим сообщения могут обрабатываться независимо друг от друга.
Тогда мы можем операцию сортировки выполнять в другом потоке. Например, запустим Воркер, передадим ему данные и завершим обработку события. И начнем обработку следующего.
Таким образом мы быстро обрабатываем события message, очередь необработанных не скапливается, но висят несколько (3-4) параллельных потока.

В этом случае, сообщение можно куда то сохранять. Не знаю, что у вас за клиент, и какого размера и вида сообщения, но на браузере можно было бы воспользоваться localStorage или indexedDB. В функции обработки события сначала сохраняем сообщение, а после окончания его обработки в Воркере удаляем. В этом случае у нас в базе при выключении клиента останутся необработанные сообщения.

[NatalliaF]

Т.е. в onmessage происходит лишь сохранение сообщений для последующей обработки воркерами? Количество воркеров определяется экспериментально (и зависит от загруженности клиента).