Проектируем распределённый планировщик задач: отказоустойчивость и масштабирование

Почему cron не справляется с миллионом задач?

Запустить задачу каждую ночь в 2 часа? Cron справится. А теперь представьте: миллион задач, каждая со своим расписанием, на сотнях машин, где любой узел может упасть прямо во время выполнения. При этом каждая задача должна выполниться ровно один раз и без опозданий. Одиночный cron здесь не помощник. Именно эту проблему решает распределённый планировщик задач, и самое интересное в нём — обработка отказов.

Основная проблема: разделение времени и выполнения

У планировщика две обязанности, которые легко перепутать: решить, когда задача должна запуститься, и собственно запустить её. Первое — проблема времени, второе — проблема выполнения. Если объединить их в один процесс, как в cron, получается просто, но хрупко. В распределённой архитектуре их разделяют.

Два жёстких требования: задача, которая должна выполниться, обязана когда-нибудь запуститься (долговечность), и задача не должна выполниться дважды, если вы этого не просили (корректность). Наивные реализации молча нарушают одно из них при первом же сбое воркера.

Ключевые проектные решения

Храните задачи в долговечном хранилище с меткой следующего запуска

Каждая задача — это запись: её расписание (cron-выражение или интервал), время следующего запуска, полезная нагрузка и статус. Задача планировщика — находить записи, где next_run уже в прошлом, и передавать их на выполнение. Если хранилище надёжно, сбой планировщика ничего не теряет, потому что источник истины — база данных, а не память.

Опрашивайте хранилище с индексом по времени, а не держите таймеры в памяти

Хранить миллион таймеров в памяти — значит потерять всё при перезапуске. Вместо этого индексируйте задачи по next_run и периодически запрашивайте те, что уже пора запускать. Интервал опроса — компромисс между задержкой и нагрузкой: опрос каждую секунду даёт запуск с точностью до секунды, но требует одного запроса в секунду.

import time
import sqlite3

def poll_due_jobs(db_path, poll_interval=1):
    conn = sqlite3.connect(db_path)
    while True:
        now = int(time.time())
        # Выбираем задачи, которые пора запускать
        due = conn.execute(
            "SELECT id, payload FROM jobs WHERE next_run <= ? AND status = 'pending'",
            (now,)
        ).fetchall()
        for job_id, payload in due:
            # Пытаемся захватить задачу (CAS)
            conn.execute(
                "UPDATE jobs SET status = 'running' WHERE id = ? AND status = 'pending'",
                (job_id,)
            )
            if conn.total_changes > 0:
                # Успешно захватили — отправляем на выполнение
                dispatch_to_worker(job_id, payload)
        time.sleep(poll_interval)

Разделите диспетчер и воркеры через очередь

Когда задача становится актуальной, диспетчер помещает её в очередь, откуда воркеры забирают. Это позволяет независимо масштабировать выполнение и поглощать всплески нагрузки. Диспетчер остаётся лёгким и занимается только временем.

from queue import Queue

dispatch_queue = Queue()

def dispatch_to_worker(job_id, payload):
    dispatch_queue.put((job_id, payload))

def worker_loop():
    while True:
        job_id, payload = dispatch_queue.get()
        execute_job(job_id, payload)
        dispatch_queue.task_done()

Обеспечьте ровно-один-раз через идемпотентность

Гарантировать доставку ровно один раз в распределённой системе невозможно, поэтому цель — как минимум один раз плюс идемпотентность выполнения. Дайте каждому запланированному запуску уникальный run_id. Перед выполнением воркер захватывает запуск условной записью (compare-and-set: меняет статус с pending на running). Если захват не удался — другой воркер уже работает. Если воркер умирает во время выполнения, таймаут видимости или истечение аренды позволяет перезахватить задачу.

import uuid

def claim_job(conn, job_id):
    run_id = str(uuid.uuid4())
    conn.execute(
        "UPDATE jobs SET status = 'running', run_id = ? WHERE id = ? AND status = 'pending'",
        (run_id, job_id)
    )
    return conn.total_changes > 0

def execute_job(job_id, payload):
    # Идемпотентное выполнение: если задача уже завершена, ничего не делаем
    if is_already_done(job_id):
        return
    # Выполняем полезную работу
    process(payload)
    mark_completed(job_id)

Избегайте стадного эффекта в популярные моменты

Все любят запускать задачи в начале часа. Если десять тысяч задач сработают одновременно, нагрузка на downstream-системы будет катастрофической. Добавьте джиттер (случайную задержку) или шардируйте временные срезы.

import random

def compute_next_run(schedule, base_time):
    # Добавляем случайный джиттер до 5% от интервала
    jitter = random.uniform(0, 0.05) * schedule.interval
    return base_time + jitter

Повторяющиеся задачи и дрейф времени

Для повторяющейся задачи после выполнения нужно вычислить следующее время запуска на основе расписания и обновить запись. Вычисляйте от планового времени, а не от текущего, иначе накопится дрейф. Если выполнение задержалось, решите политику: пропустить пропущенный запуск или выполнить его немедленно. Оба варианта допустимы, но выбирайте осознанно.

Как устроены реальные системы

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

Ментальная модель для собеседования: время и выполнение — разные проблемы, долговечность обеспечивается хранилищем, а не процессом, и ровно один раз — это на самом деле как минимум один раз плюс идемпотентные захваты.

Практический вывод

Попробуйте спроектировать свой планировщик, используя описанные принципы. Начните с простого: SQLite, очередь в памяти, один воркер. Затем добавьте отказоустойчивость: несколько воркеров, аренду с таймаутом, идемпотентность. Вы увидите, как каждое решение влияет на гарантии.