Когда я работал в Reddit и отвечал там за инфраструктуру, самой важной поддерживаемой системой для меня была Postgres, а на втором месте стоял брокер сообщений RabbitMQ. Он был необходим для работы Reddit — перед сохранением в базу данных все данные поступали в распределённую очередь. Например, если пользователь лайкал пост, то это записывалось в очередь и кэш, а затем пользователю передавалось сообщение об успешном выполнении. Затем программа обработки очереди брала этот элемент и пыталась записать его в базу данных, а также создать новую рабочую операцию для пересчёта всех списков, на которые влияет этот лайк.

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

• Горизонтальной масштабируемостью. Очереди задач позволяют параллельно выполнять множество задач, используя ресурсы множества серверов. Масштабировать систему было довольно просто — достаточно было добавлять новых воркеров.

• Управление потоками. Работая с очередями задач, мы могли управлять частотой потребления воркерами задач из различных очередей. Например, если задача требовала больших ресурсов, мы могли ограничивать количество таких конкурентно выполняемых задач в одном воркере. Если задача получала доступ к API с ограничением частоты использования, то мы могли ограничивать количество выполняемых в секунду задач, чтобы не перегружать API. 

• Планирование. Очереди задач позволяют определять, когда и как часто выполняется задача. Например, мы можем запускать задачи в планировщике cron или планировать исполнение задач в какой-то момент в будущем.

Такая система хорошо масштабировалась, но могла ломаться разными хитрыми способами. Если базы данных голосования за посты и комментарии были недоступны, элемент возвращался обратно в очередь. Если кэш списков был недоступен, списки невозможно было пересчитать. Если обработчик очередей вылетал после получения элемента, но до выполнения действий с ним, то данные просто терялись. А если была недоступна сама очередь, мы могли терять голоса или комментарии или посты (У вас на Reddit когда-нибудь были мысли «я же точно голосовал здесь, но теперь голоса нет!»? Теперь вы знаете причину).

Для обеспечения надёжности распределённых очередей задач нам нужны были устойчивые очереди (durable queue), заносящие статус помещённых в очереди задач в надёжное хранилище наподобие Postgres. При наличии устойчивых очередей мы могли бы продолжать окончившиеся сбоем задачи с последнего завершённого этапа, не теряя при этом данные в случае вылета программ.

Когда я работал в Reddit, устойчивые очереди были редкостью, но сегодня они всё больше набирают популярность. По сути, они комбинируют очереди задач с устойчивыми рабочими процессами, позволяя надёжным образом управлять рабочими процессами множества параллельных задач. Архитектурно устойчивые очереди очень похожи на обычные очереди, но используют постоянное хранилище (обычно реляционную базу данных) в качестве и брокера сообщений, и бэкенда:

Базовая абстракция устойчивых очередей — это рабочий процесс (workflow) множества задач. Например, можно передать задачу обработки документа, которая разбивает документ на страницы, обрабатывает каждую страницу параллельно в отдельных задачах, а затем выполняет постобработку и возвращает результаты:

@workflow()
def process_tasks(tasks):
  task_handles = []
  # Заносим каждую задачу в очередь, чтобы все задачи обрабатывались конкурентно.
  for task in tasks:
    handle = queue.enqueue(process_task, task) task_handles.append(handle)
  # Ждём, пока каждая задача выполнится и получит результат.
  # Возвращаем результаты всех задач.
  return [handle.get_result() for handle in task_handles]

Устойчивые очереди создают чек-поинты рабочих процессов в своём постоянном хранилище. Когда клиент передаёт задачу, задача и входные данные записываются. Когда эта задача вызывает другую задачу, эта подзадача и её входные данные записываются, как дочерний элемент вызвавшей их задачи. Таким образом, система очередей имеет полное постоянное хранилище всех задач и их взаимосвязей.

Эти рабочие процессы важнее всего при восстановлении после сбоев. В случае, когда при выполнении задачи работа неустойчивого воркера прерывается, очередь в лучшем случае перезапускает его с начала, а в худшем теряет задачу. Это не подходит для длительно выполняемых рабочих процессов и задач с критичными данными. Когда устойчивая система очередей восстанавливает рабочий процесс, она проверяет его чек-поинты, чтобы восстановиться из последнего завершённого этапа, избегая таким образом повторной передачи завершённой работы.‍

Устойчивые очереди и наблюдаемость

Ещё одно преимущество устойчивых очередей — встроенная наблюдаемость. Так как эти очереди сохраняют подробные записи о каждом переданном рабочем процессе и задаче, они упрощают мониторинг того, что делают эти очереди и рабочие процессы в любой момент времени. Например, для изучения текущего содержимого очереди (или содержимого в прошлом) достаточно лишь SQL-запроса. Аналогично, для изучения текущего состояния рабочего процесса тоже нужен ещё один SQL-запрос.‍

Минусы устойчивых очередей

Когда же использовать устойчивые очереди? Как всегда, это зависит от компромиссов. Для устойчивых очередей главный компромисс — это производительность брокера сообщений. Большинство распределённых очередей задач для брокеринга сообщений и хранения выходных данных задач использует хранилище ключей и значений в памяти наподобие Redis. Однако устойчивым очередям нужно использовать в качестве брокера сообщений и бэкенда устойчивое хранилище; часто это реляционная база данных наподобие Postgres. Последняя обеспечивает более надёжные гарантии, но ценой снижения пропускной способности. Таким образом, устойчивые очереди следует выбирать в случае обработки малых объёмов крупных, критичных для бизнеса задач, а распределённые очереди задач — при обработке очень большого объёма мелких задач.‍

Дополнительные источники

• Dosu — миграция очередей с Celery на DBOS

• Bristol Myers Squibb — устойчивое горизонтальное масштабирование конвейеров геномных данных при помощи DBOS

• cStructure — миграция очередей с Celery на DBOS

• DOCS: DBOS durable queuing