Использование GenStage/Flow для обработки событий в реальном времени

Question

В настоящее время я создаю систему обработки событий в реальном времени с использованием Elixir, и сейчас я пытаюсь обернуть голову вокруг GenStage/Flow, чтобы понять, являются ли это правильные абстракции, на которые можно опираться. К сожалению, примеры более крупных приложений немногочисленны, и большинство из них касается параллельной обработки устаревших данных. Я использую бесконечный поток событий в качестве источника.

Мой вопрос заключается в том, имеет ли смысл использовать GenStage/Flow для случая, когда я подписываюсь на бесконечный внешний поток событий, который отправляет события в мое приложение. Я хочу, чтобы события обрабатывались сразу по прибытии на сервер. То есть я не хочу их буферизовать, пока не получу 500, чтобы Flow запустил запрос. Но имеет ли смысл использовать минимальное требование 1?

Answer 1

Я бы сказал, что GenStage практически не работает в режиме реального времени. Основная идея заключается в том, чтобы отложить оценку до тех пор, пока не появится спрос (скажем, работник готов). Если вы хотите обработать их сразу, просто создайте новый процесс Elixir для каждого события и молитесь, чтобы планировщик не задохнулся:)

Даже если вы используете минимальное требование 1, оно будет обрабатываться последовательно, и вы будете получать буферизированные события переполнения. Да, вы можете распараллелить, но я не уверен, придется ли вам начинать параллельные этапы сразу или нет. И снова, когда вы достигнете N одновременных событий с N параллельными рабочими, события будут буферизованы.

ОБНОВЛЕНИЕ После небольшого рассмотрения я думаю, что в некоторых случаях GenStage или его производный поток более высокого порядка все еще могут быть полезны для обработки почти в реальном времени. Чтобы избежать накладных расходов при создании процессов, можно использовать окна фиксированной ширины для сбора и разделения событий, которые затем могут обрабатываться различными потребителями, пулами или даже отдельными процессами, если это необходимо. Единственный недостаток или, скорее, ограничение - это введение квантования, которое может составлять всего 1 мс: https://github.com/elixir-lang/flow/blob/v0.14.2/lib/flow/window.ex#L324 Однако это всего лишь теоретическое предположение, я не пробовал это.