Периодические задачи по cron

Отчёт по продажам — каждый день в 06:00, пересчёт рейтингов — каждую ночь в 03:00, дайджест новостей — по будням в девять. ProcessIn из пятого урока здесь не помощник: он откладывает задачу один раз, а вам нужен повтор по календарю, без конца. Можно вынести расписание в классический cron системы и ставить задачи шелл-скриптом, но тогда часть логики уезжает из кода. Asynq предлагает встроенный ответ: Scheduler — планировщик, который живёт в вашем процессе и в моменты тика сам ставит задачи в очередь через Redis. Ключевая мысль, которая экономит час чтения документации: планировщик ничего не исполняет. Он только постановщик — после тика в очереди лежит обычная задача, и её подхватывают обычные воркеры с обычными ретраями и очередями. В уроке — регистрация расписания, таймзоны, динамический список задач через PeriodicTaskManager и главный подводный камень периодики: что делать, когда задача выполняется дольше, чем длится интервал. И одна деталь топологии, к которой вернёмся с кодом: планировщик — отдельная роль в деплое, и при масштабировании воркеров она не должна плодить копии автоматически.

Asynq cron: Scheduler и Register#

Asynq cron начинается с двух строк: scheduler := asynq.NewScheduler(redisConnOpt, opts) и Register. Метод Register принимает cron-выражение, задачу и любые опции постановки — те же, что у Enqueue: очередь, MaxRetry, Unique — и возвращает идентификатор записи. По этому идентификатору расписание можно снять методом Unregister, когда задача перестала быть нужна. Формат выражений знакомый: пять полей «минуты часы день-месяца месяц день-недели» вроде 0 6 *, плюс интервальные записи @every 30s и @every 1h для случаев, когда привязка к часам суток не нужна. Один планировщик держит сколько угодно записей: Register возвращает новый идентификатор на каждый вызов, и повторная регистрация того же выражения ошибкой не считается — дубликатом она становится уже в рантайме, когда два тика ставят две задачи. Невалидное выражение, напротив, ловится сразу: Register разбирает его и возвращает ошибку, не добавляя запись. Запускать планировщик можно двумя способами: Run блокирует выполнение до сигнала остановки — отдельный процесс-планировщик, — а Start возвращает управление сразу, если планировщик встраивается в существующий сервис, где блокироваться нельзя. Каждая постановка по тику проходит через Redis как обычная клиентская: если Redis в момент тика недоступен, планировщик не потеряет молча — ошибку можно обработать через EnqueueErrorHandler в опциях.

Планировщик с двумя расписаниями

loc, err := time.LoadLocation("Europe/Moscow")
if err != nil {
    log.Fatal(err)
}

scheduler := asynq.NewScheduler(redisConnOpt, &asynq.SchedulerOpts{
    Location: loc,
})

task := asynq.NewTask(tasks.TypeDailyReport, nil)
entryID, err := scheduler.Register("0 6 * * *", task, asynq.Queue("reports"))
if err != nil {
    log.Fatal(err)
}
log.Printf("запись расписания: %s", entryID)

scheduler.Run()

Register принимает cron-выражение, задачу и опции постановки — здесь очередь reports. Возвращённый entryID пригодится для Unregister, когда расписание перестанет быть нужным. Run блокируется до сигнала остановки, поэтому такой main-код живёт в отдельном процессе; для встройки в сервис есть неблокирующий Start. Location решает, какое именно 06:00 имеется в виду.

Cron-выражения и таймзона по умолчанию#

Выражение из пяти полей читается слева направо: минуты, часы, день месяца, месяц, день недели; звёздочка — «любое значение», дефис — диапазон, запятая — перечисление, слэш — шаг. Если синтаксис вам незнаком целиком, у сайта есть отдельный курс по cron — здесь остановимся на главном сюрпризе. По умолчанию SchedulerOpts.Location не задана, и расписание живёт в UTC: выражение 0 6 * сработает в шесть утра по Гринвичу, то есть в девять по Москве — и почти каждая команда ловит это на первом деплое. Лечение две строки: LoadLocation в инициализации и Location в SchedulerOpts, как в коде выше. Проверка после деплоя бесплатная: планировщик печатает свою таймзону в лог при старте, так что строка «Scheduler timezone is set to ...» — быстрый способ убедиться, что вы в правильной зоне. Для минимальных Docker-образов напоминание из пятого урока остаётся в силе: без tzdata LoadLocation вернёт ошибку, лечится пакетом tzdata или импортом time/tzdata. Отдельно про @every: он не привязан к часам суток и отсчитывает интервал от запуска планировщика, поэтому после рестарта в 10:47 тики пойдут в 11:47, 12:47 — удобно для «раз в час», но не для «каждый день в шесть».

Проверка: таймзона по умолчанию

Scheduler создан с SchedulerOpts по умолчанию, выражение 0 6 * * *. Во сколько по Москве (UTC+3) сработает первая постановка?

Кто ставит задачу при тике#

В момент тика планировщик делает постановку через Redis — и дальше задача живёт своей обычной жизнью: попадает в очередь, забирается воркером, при сбое повторяется с экспонентой из шестого урока. Для задач в расписании работают и опции постановки: очередь из Register определяет класс обслуживания, MaxRetry задаёт потолок повторов. У SchedulerOpts есть и хуки: PreEnqueueFunc вызывается перед каждой постановкой по тику, PostEnqueueFunc — после, уже с TaskInfo и ошибкой на руках; через них удобно считать реальные постановки и отдавать метрику «тик случился». EnqueueErrorHandler из ранних версий делает то же для сбоев и помечен устаревшим. Из этой механики следует два неудобных факта, о которых лучше знать заранее. Первый: пропущенный тик не догоняется — если планировщик стоял в 06:00, постановки не было, и после подъёма задача не появится задним числом; расписание смотрит только вперёд. Второй: планировщики не координируются между процессами — asynq не держит лидера и не проверяет, кто уже зарегистрировал это выражение. Два живых планировщика с одинаковым Register ставят задачу дважды в каждый тик. Поэтому правило топологии простое: на каждое расписание — ровно один живой планировщик в кластере, и этому правилу посвящён коллаут ниже.

Живое место: две реплики — два письма

Воркеры масштабируются репликами — а в main.go одной из реплик остался тот же Register, что и у первой. Итог: каждое утро дайджест уходит пользователям дважды, половина подписчиков отписывается, а в логах всё чисто — обе постановки честно успешны. Планировщики не координируются между собой, Redis-замков нет. Решения: выделенный однореплиочный процесс-планировщик, выбор лидера между репликами или вынос регистрации за флаг роли в деплое.

Расписания видны в дашборде

Планировщики пишут в Redis записи о себе, и asynqmon показывает их на вкладке Scheduler Entries: какие выражения зарегистрированы, кем и когда был предыдущий тик. Отсюда дешёвые проверки после деплоя: ровно одна запись на каждое расписание, а не две, и записи живы — значит, процесс-планировщик жив. Мёртвый планировщик пропадает из списка через несколько десятков секунд после остановки: записи пишутся с коротким сроком жизни.

Периодические задачи asynq: PeriodicTaskManager#

Пока периодических задач пять, Register в инициализации — самый честный код. Когда их десятки и список меняется — расписание выгружается из базы, добавляется клиентами, отличается по окружениям, — в игру вступают периодические задачи asynq в динамическом режиме: PeriodicTaskManager. Менеджер создаётся через asynq.NewPeriodicTaskManager с опциями PeriodicTaskManagerOpts: в них передаётся PeriodicTaskConfigProvider — ваш источник конфигов, RedisConnOpt для соединения и вложенные SchedulerOpts. Провайдер реализует интерфейс с одним методом GetConfigs, который возвращает срез значений PeriodicTaskConfig — это структура с полями Cronspec, Task и опциональными Opts. Раз в SyncInterval — по умолчанию три минуты — менеджер опрашивает провайдера, сравнивает полученный список с зарегистрированными записями по хэшу «выражение, тип задачи, payload, опции» и сам доводит реальность до желаемого: новые регистрирует, исчезнувшие снимает. Ваш код меняет строки в базе — планировщик меняет расписание без перезапуска. Детали, которые экономят отладку: первая синхронизация происходит сразу при Start, до старта планировщика, — расписание из базы подхватывается без ожидания трёх минут; вместо RedisConnOpt менеджеру можно отдать готовый универсальный клиент Redis, если он уже заведён для других нужд.

Провайдер конфигов и менеджер

type DBProvider struct{ DB *sql.DB }

func (p *DBProvider) GetConfigs() ([]*asynq.PeriodicTaskConfig, error) {
    // читаем cronspec и тип из таблицы periodic_jobs
    return []*asynq.PeriodicTaskConfig{
        {Cronspec: "0 3 * * *", Task: asynq.NewTask(tasks.TypeRecalc, nil)},
    }, nil
}

mgr, err := asynq.NewPeriodicTaskManager(asynq.PeriodicTaskManagerOpts{
    RedisConnOpt:               redisConnOpt,
    PeriodicTaskConfigProvider: &DBProvider{DB: db},
})
if err != nil {
    log.Fatal(err)
}

mgr.Start()

Провайдер реализует PeriodicTaskConfigProvider: метод GetConfigs возвращает срез PeriodicTaskConfig — структур с Cronspec, Task и опциональными Opts. Менеджер раз в SyncInterval опрашивает провайдера, сравнивает конфиги по хэшу «выражение, тип, payload, опции» и сам регистрирует новые записи и снимает исчезнувшие. Start запускает и планировщик, и цикл синхронизации; Shutdown останавливает оба.

Проверка: опечатка в выражении

Scheduler.Register вызвали с опечаткой в выражении — 0 6* * * без пробела после шестёрки. Что произойдёт?

Идемпотентность периодических задач#

Каждый тик ставит задачу независимо от прошлого: планировщик не проверяет, кончился ли ночной пересчёт, упала ли вчерашняя выгрузка, держится ли блокировка. Отсюда главный эксплуатационный навык темы — хендлер периодической задачи обязан быть идемпотентным и уметь переживать наложение смен: прошлый прогон ещё идёт, а новый уже поставлен. Три приёма уровня хендлера: проверка «не выполняется ли уже» по служебной записи в базе или Redis; деление работы на окна, чтобы каждый прогон обрабатывал свой кусок и пропускал чужой; отказ от догоняющей логики — пропущенный пересчёт не важнее сегодняшнего. Если наложение принципиально недопустимо, поможет фиксированный идентификатор задачи и опция Unique с TTL на интервал: постановка с живым дублем вернёт ErrTaskIDConflict, механику которого разбирает двенадцатый урок. Полная дисциплина идемпотентных хендлеров — своя территория восемнадцатого урока; здесь достаточно правила: cron ставит задачу, не глядя на прошлое, — значит, хендлер смотрит на прошлое сам. Практическая проверка идемпотентности простая: запустите хендлер дважды на одних данных — второй прогон должен либо ничего не сделать, либо проделать то же без побочных эффектов; всё остальное — баг, который cron однажды превратит в двойную отправку.

Когда приедет первая постановка?

Прочитайте код и выберите, что он выведет:

// планировщик стартовал в 10:47
entryID, err := scheduler.Register("0 6 * * *", reportTask)

Кандидаты в периодические задачи#

Ночные пересчёты: рейтинги, агрегаты, тёплые кэши к началу рабочего дня.
Дайджесты и сводки: письмо по утрам, выгрузка раз в час, сводка по будням.
Синхронизация с внешними системами: каталог поставщика раз в полчаса.
Очистка: устаревшие сессии, временные файлы, забытые черновики.
Проверки здоровья: ping зависимостей с постановкой алерт-задачи.
Календарные напоминания: вынести дедлайн из данных в расписание не выйдет — для событий остаётся ProcessAt.

Разовая отложка против периодики#

Инструмент	Кто ставит	Повтор	Кейс
ProcessIn / ProcessAt	ваш код на событии	один раз	напоминание за час до приёма
Scheduler	планировщик по тику	по cron-выражению	отчёт в 06:00 ежедневно
PeriodicTaskManager	менеджер по провайдеру	по cron-выражению	расписание живёт в базе
классический cron	планировщик ОС	по cron-выражению	постановка из шелла без Go

Соберите регистрацию расписания

Допишите конструктор планировщика, структуру опций с таймзоной и метод регистрации выражения.

scheduler := asynq.(redisConnOpt, &asynq.{ Location: loc, }) entryID, err := scheduler.("0 6 * * *", reportTask)

Соберите процесс-планировщик

Расставьте строки в порядке работы: таймзона, планировщик, задача, регистрация, запуск.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: периодические задачи

Термин

Scheduler

нажмите, чтобы перевернуть

Определение

Планировщик asynq: живёт в вашем процессе и по тикам cron-выражений ставит задачи в Redis. Только постановщик — исполняют обычные воркеры.

1 / 6

Периодика собрана: Scheduler регистрирует выражения и ставит задачи по тикам, Location держит таймзону под контролем, PeriodicTaskManager синхронизирует расписание с вашим источником конфигов, а идемпотентность хендлера закрывает наложения смен. Мини-эксперимент: поднимите процесс-планировщик с выражением на минуту вперёд и остановите воркеры — постановка случится, а задача дождётся в очереди, как и любая другая; затем убейте планировщик на следующий тик и убедитесь, что постановки не было: пропущенный тик не догоняется. С той же схемой проверьте таймзону: выражение 0 6 * без Location сработает по UTC, и лог планировщика честно это покажет. Формат прежний: локальный Go и Redis в Docker, песочница сайта — тематическая программа на чистом Go. Дальше по курсу: десятый урок покажет противоположный приём — не ставить задачи по расписанию, а собирать уже поставленные в пачки; разовые отложки ProcessIn из пятого урока остаются вашим инструментом для событий, а cron — для календаря.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Scheduler регистрирует cron-выражения: Register(spec, task, opts...) возвращает entryID для Unregister.
Планировщик только ставит задачи: исполнение — обычные воркеры с очередями и ретраями.
Одно расписание — один живой планировщик на кластер: два Register дают двойные постановки.
Cron по умолчанию живёт в UTC: таймзону задаёт SchedulerOpts.Location через LoadLocation.
Пропущенный тик не догоняется: не было планировщика — не было постановки.
PeriodicTaskManager синхронизирует расписание с провайдером GetConfigs раз в SyncInterval.
Конфиг периодической задачи — структура PeriodicTaskConfig: Cronspec, Task, Opts.
Хендлер периодической задачи обязан быть идемпотентным: наложение смен — норма жизни.

Домашнее задание

Соберите отдельный процесс-планировщик с двумя задачами: отчёт по выражению 0 6 * * * с Location Europe/Moscow и пересчёт @every 1h с очередью low. Убедитесь в asynqmon, что постановки появляются как обычные задачи в нужной очереди, и что воркеры исполняют их с обычными ретраями. Затем поднимите вторую копию планировщика и посчитайте дубликаты за день — почините, оставив одну реплику, и зафиксируйте правило в README сервиса. Всё локально: Go и Redis из Docker; песочница сайта сетевой код не исполняет.

Вопросы и ответы по уроку

Scheduler или PeriodicTaskManager — что выбрать?

Пять статичных расписаний — Register в инициализации: код прозрачен, расписание видно в одном файле. Десятки задач, изменения без деплоя, расписание в базе — PeriodicTaskManager с провайдером: он сам добавляет и снимает записи по списку от GetConfigs. Менеджер внутри использует тот же Scheduler, так что миграция между ними дешёвая. Промежуточная точка — Register в инициализации плюс хранение выражений в базе на будущее: статичные расписания не требуют инфраструктуры менеджера, а переезд останется делом одного коммита.

Что случится с расписанием при падении процесса-планировщика?

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

Можно ли ставить периодическую задачу в конкретную очередь с ретраями?

Да: Register принимает те же опции, что и Enqueue — asynq.Queue("reports") отправит постановки в нужную очередь, MaxRetry ограничит повторы при сбоях, TaskID даст фиксированный идентификатор. Постановка из планировщика ничем не отличается от клиентской — просто автором выступает ваш процесс.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.