--- title: "Итоговый проект: система рассылок" description: "Итоговый asynq проект: система email-рассылок. Отложенные письма, агрегация, ретраи, дашборд. Welcome с отменой, дайджест группами, graceful shutdown." course: "Asynq в Go: очередь фоновых задач на Redis" url: "https://golangmaster.ru/courses/asynq/lessons/25-itogovyj-proekt/" published: "2026-09-19" updated: "2026-09-23" minutes: 25 --- # Итоговый проект: система рассылок *Курс «Asynq в Go: очередь фоновых задач на Redis», урок 25 из 25. ~25 минут. Обновлено 23.09.2026. Источник: https://golangmaster.ru/courses/asynq/lessons/25-itogovyj-proekt/* Финальный проект собирает курс в рабочую систему email-рассылок: подписка ставит отложенное welcome-письмо, отписка успевает его отменить, периодический дайджест собирается группами, а дашборд показывает всё это вживую. Новых API почти нет — проект проверяет умение соединять знакомые детали в надёжный конвейер. Продукт простой и узнаваемый: человек оставляет почту — получает welcome-письмо, а раз в неделю — дайджест новинок. За простотой стоит конвейер из задач, который обязан не терять письма, не дублировать их и не душить почтовый шлюз. Два типа задач: welcome и digest. Welcome ставится из HTTP-хендлера с отложкой в пять минут — за это окно человек успевает передумать, и письмо честно не уходит. Дайджест — периодическая задача, а уведомления внутри неё собираются в пачки агрегатором, чтобы один пользователь получил одно письмо, а не пятьдесят. Под письмами — два слоя инфраструктуры: очереди critical и default с разными приоритетами и middleware с логами и метриками, через который проходит каждая задача. Сборка идёт по швам, которые уже разобраны: постановка из веба, идемпотентность по ID, группы, ретраи, тесты, остановка. Каждое решение помечено номером урока-источника, так что непонятный стык открывается заново за минуту, а весь каркас переносится в свой проект почти без правок. Порядок разбора: сначала карта проекта и общая архитектура, затем welcome-конвейер с постановкой из веба и двойной защитой отписки, дальше дайджест с периодикой и агрегацией, и в конце — надёжность: ретраи, тесты, наблюдаемость и остановка. Строки кода сокращены до существенного: декларации типов, обвязка логирования и импорты опущены, логика сохранена целиком. ### Архитектура: два типа задач и две очереди Проект живёт в трёх частях: HTTP-сервис принимает подписки и отписки, клиент ставит задачи, сервер их исполняет — все три могут быть одним процессом на старте и разными сервисами в проде. Типы задач задаются константами: TypeWelcomeEmail и TypeDigestEmail регистрируются в mux из восьмого урока. Очереди две: welcome-письма едут в default, дайджест — в critical, потому что утренняя рассылка не должна ждать за потоком одиночных писем; приоритеты и ловушки нескольких очередей — тема седьмого урока. Вокруг каждой задачи — middleware из двадцать третьего урока: лог с типом, ID и длительностью, счётчики обработок и ошибок; без него рассылка — чёрный ящик, с ним — наблюдаемый процесс. Отдельный закон проекта: письма шлёт только воркер, никакой хендлер не отправляет почту синхронно. HTTP-ответ возвращает 202 Accepted сразу после постановки задачи — пользователь не ждёт SMTP, а очередь гарантирует доставку даже при перезапуске сервиса. Клиент и сервер строятся на одном RedisClientOpt и живут в одном репозитории, но пакеты раздельные: api, tasks, mailer — при разделе на два сервиса граница уже проведена. Регистрация хендлеров идёт через srv.Handle с константой типа: опечатка в имени ловится компиляцией, а не ночным инцидентом. Гонки между подпиской и отпиской разруливают два механизма: детерминированный ID задачи и проверка состояния подписки в самом хендлере. ### Кирпич проекта — урок-источник | Кирпич | Урок | Роль в проекте | | --- | --- | --- | | ProcessIn(5 * time.Minute) | 05 | окно на отмену: письмо уходит не сразу | | TaskID("welcome:" + email) | 04, 12 | детерминированный ID, дубль постановки ловится | | Очереди critical и default | 07 | дайджест не ждёт за одиночными письмами | | GroupAggregator + GroupGracePeriod | 10 | уведомления одного пользователя — одной пачкой | | Ретраи с backoff, SkipRetry | 06, 13 | SMTP деградирует — письма не теряются, битые не зацикливаются | | Тесты хендлеров без Redis | 22 | логика писем держится под замком | | Graceful shutdown | 19 | деплой не роняет письма на полуслове | ### Проверка: состояние после ProcessIn **Вопрос:** Welcome-письмо поставлено через asynq.ProcessIn(5 * time.Minute). Где задача через минуту? - В состоянии pending: очередь сразу взяла её в работу - ✅ В состоянии scheduled: исполнится, когда истекут пять минут - В состоянии active: воркер уже держит задачу - В состоянии retry: постановка с ProcessIn начинается с ретрая ProcessIn откладывает исполнение: задача попадает в scheduled и переходит в pending только к моменту срока. Именно поэтому окно в пять минут — реальное время на отмену: пока задача scheduled, её можно удалить из очереди. ### Welcome: постановка из HTTP-хендлера Хендлер подписки делает четыре шага. Первый — декод тела и нормализация почты: TrimSpace и нижний регистр, иначе ada@example.com и ADA@example.com станут двумя подписчиками. Второй — сохранение подписчика в хранилище: если запись уже есть, это доменная ситуация, отвечаем 409, а не ставим второе письмо. Третий — payload: json.Marshal структуры WelcomePayload с одним полем Email; payloads из четвёртого урока держат маленькими и плоскими. Четвёртый — постановка: asynq.NewTask(TypeWelcomeEmail, payload) и client.Enqueue с тремя опциями. ProcessIn(5 * time.Minute) даёт окно на отмену — тарифицированное письмо не уходит мгновенно. TaskID("welcome:" + email) делает постановку идемпотентной: повторная подписка на ту же почту вернёт ошибку дубликата вместо второго письма — приём из уроков про ID и уникальность. Queue("default") и MaxRetry(5) завершают контракт. Ошибку дубликата хендлер распознаёт через errors.Is(asynq.ErrDuplicateTask) и трактует спокойно: подписка уже активна, клиенту достаточно 202 — но лог запишет попытку, потому что молчаливый дубль однажды окажется не дублем, а симптомом. Для постановки с наследованием контекста запроса есть EnqueueContext: пока подписка ставится синхронно и быстро, обычного Enqueue хватает, но при постановке из под веба при таймаутах клиента контекстная версия честнее. ### POST /subscribe: постановка welcome ```go func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request) { var req SubscribeRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "битое тело", http.StatusBadRequest) return } email := strings.ToLower(strings.TrimSpace(req.Email)) if err := s.store.CreateSubscriber(r.Context(), email); err != nil { http.Error(w, "уже подписан", http.StatusConflict) return } payload, err := json.Marshal(WelcomePayload{Email: email}) if err != nil { http.Error(w, "marshal", http.StatusInternalServerError) return } _, err = s.client.Enqueue(asynq.NewTask(TypeWelcomeEmail, payload), asynq.ProcessIn(5*time.Minute), asynq.TaskID("welcome:"+email), asynq.Queue("default"), asynq.MaxRetry(5), ) if err != nil { log.Printf("enqueue welcome %s: %v", email, err) } w.WriteHeader(http.StatusAccepted) } ``` Хендлер не отправляет почту — он фиксирует намерение в очереди и отвечает 202. ProcessIn откладывает письмо на пять минут, TaskID строит детерминированный идентификатор из адреса: повторный POST на ту же почту упадёт с ошибкой дубликата, не создав вторую задачу. Ошибка постановки не роняет ответ — письмо уже сохранено в хранилище, а провал логируется для ручного повтора. ### Welcome-хендлер: идемпотентность и отписка Обратная сторона отложки — за пять минут человек может передумать, и письмо обязано это пережить. Отписка работает в два слоя. Первый — быстрая отмена: хендлер POST /unsubscribe помечает подписку неактивной в хранилище и просит Inspector удалить задачу DeleteTask("default", id) — для scheduled-задачи это штатная операция, а состояние подписки в базе меняется всегда, даже если задачи в очереди уже нет. Второй слой — проверка в самом welcome-хендлере: перед отправкой он перечитывает подписчика и молча выходит с nil, если подписка неактивна или не подтверждена. Двойная защита не паранойя, а норма: удаление из очереди и чтение из хранилища — две разные системы, и гонка «письмо уже в active, отписка приходит в этот момент» разрешается только проверкой на месте. Хендлер идемпотентен и по-другому: он не держит в голове, сколько раз его вызовут, — повторный вызов после ретрая упадёт на том же фильтре и не отправит второе письмо. Таймауты из одиннадцатого урока заданы опцией Timeout при постановке: письмо обязано уйти за две минуты, иначе попытка проваливается и очередь забирает её снова — висячий SMTP-коннект не удержит воркер навсегда. Ошибки классифицируются по четырнадцатому уроку: битый payload — SkipRetry, недоступность хранилища и SMTP — обычный возврат ошибки с wrap, дальше работает backoff из шестого. ### Хендлер welcome: проверка перед отправкой ```go func (s *Server) handleWelcome(ctx context.Context, t *asynq.Task) error { var p WelcomePayload if err := json.Unmarshal(t.Payload(), &p); err != nil { return fmt.Errorf("unmarshal welcome: %v: %w", err, asynq.SkipRetry) } sub, err := s.store.Subscriber(ctx, p.Email) if err != nil { return fmt.Errorf("load subscriber %s: %w", p.Email, err) } if sub == nil || !sub.Confirmed { log.Printf("welcome skip: %s отписался", p.Email) return nil } if err := s.mailer.SendWelcome(ctx, p.Email); err != nil { return fmt.Errorf("send welcome %s: %w", p.Email, err) } log.Printf("welcome sent: %s", p.Email) return nil } ``` Хендлер перечитывает состояние подписки непосредственно перед отправкой — это второй слой защиты, который закрывает гонку с отпиской. Битый payload получает SkipRetry: ретраить нечего, данные не исправятся. Хранилище и SMTP упавшими ошибками с wrap уходят в обычный ретрай по политике, а сам хендлер остаётся чистой функцией — с fake-хранилищем и fake-mailer он тестируется без Redis. > **Живое место: гонка отписки и письма.** Самый скользкий стык проекта. Отписка удалила задачу — а письмо уже в active на другом воркере: DeleteTask не трогает исполняемые задачи, и единственный барьер остаётся в хендлере. Обратная гонка тоже возможна: проверка подписки прошла, письмо уходит — и в этот же миг человек жмёт отписаться. Идеала нет, есть договор: письмо, отправленное в окне пары секунд вокруг отписки, считается допустимым, а вот второе welcome-письмо — нет. Поэтому TaskID в постановке обязателен: он отсекает дубль на входе, а не на выходе, когда письмо уже ушло. ### Проверка: отмена отложенного письма **Вопрос:** Человек отписался через две минуты после подписки, welcome-задача ещё в scheduled. Что делает отписка? - Ничего: scheduled-задачу отменить нельзя, письмо уйдёт - ✅ inspector.DeleteTask("default", id): scheduled-задачу можно удалить - inspector.CancelProcessing(id): сигнал отмены контекста - UpdateTaskPayload с пустым адресом: письмо не отправится DeleteTask работает для pending, scheduled, retry и archived — scheduled-письмо удаляется штатно. CancelProcessing адресован только исполняемым задачам, а UpdateTaskPayload меняет содержимое, но не отменяет отправку. Проверка в хендлере остаётся вторым слоем на случай гонки. ### Дайджест: периодика и группы Периодический дайджест ставится двумя способами, и оба уже были в курсе. Способ первый — cron-постановка из девятого урока: планировщик раз в сутки парсит cron-выражение и вызывает Enqueue с ProcessAt на следующий слот. Способ второй — PeriodicTaskManager: конструктор NewPeriodicTaskManager получает RedisConnOpt и SchedulerOpts, провайдер отдаёт срез PeriodicTaskConfig с CronSpec и задачей, Run блокируется до сигнала — менеджер удобен, когда конфигурация приезжает из хранилища и меняется без деплоя. Вторая половина дайджеста — агрегация из [десятого урока](/courses/asynq/lessons/10-grupy-agregaciya/): каждое событие продукта ставит крошечную задачу с опцией Group и ключом-почтой, сервер копит их в группе, а GroupAggregatorFunc собирает пачку в один TypeDigestEmail с адресами внутри. Настройки Config управляют терпением: GroupGracePeriod — минимальное окно ожидания хвоста (меньше секунды запрещено), GroupMaxDelay — жёсткий потолок ожидания, GroupMaxSize — размер пачки, по достижении которого сборка происходит немедленно. Дайджест-хендлер получает payload со срезом адресов, шлёт одно письмо на человека и помечает события доставленными — идемпотентность и здесь держится на детерминированных ID и проверках состояния: повторный запуск пачки после падения воркера не задвоит письма, потому что отметка о доставке пишется тем же хранилищем, которое читает фильтр. ### Конфиг сервера: очереди и агрегатор ```go srv := asynq.NewServer(redisOpt, asynq.Config{ Concurrency: 10, Queues: map[string]int{ "critical": 6, "default": 3, }, GroupGracePeriod: 2 * time.Minute, GroupMaxDelay: 15 * time.Minute, GroupMaxSize: 50, GroupAggregator: asynq.GroupAggregatorFunc( func(group string, tasks []*asynq.Task) *asynq.Task { var emails []string for _, t := range tasks { var p EventPayload if json.Unmarshal(t.Payload(), &p) == nil { emails = append(emails, p.Email) } } payload, _ := json.Marshal(DigestPayload{Emails: emails}) return asynq.NewTask(TypeDigestEmail, payload) }, ), }) ``` Queues задаёт веса очередей: critical с дайджестом получает вдвое больше внимания воркеров, чем default с welcome-письмами. GroupGracePeriod держит группу открытой две минуты в надежде на хвост событий, GroupMaxDelay не даст ждать дольше пятнадцати, GroupMaxSize собирает пачку сразу при пятидесяти задачах. Агрегатор склеивает payload в один дайджест-задачу — это и есть то самое одно-письмо-на-пользователя. ### Пачка собралась раньше срока ```go // GroupGracePeriod: 2 * time.Minute // GroupMaxSize: 50 // в группу прилетела 50-я задача, // grace period ещё не истёк ``` **Что выведет программа?** - Группа дождётся конца grace period в любом случае - ✅ Агрегатор соберёт пачку немедленно: GroupMaxSize сокращает ожидание - Задачи уйдут в retry и соберутся заново - Пачка отложится ещё на GroupMaxDelay GroupMaxSize — триггер немедленной сборки: как только в группе набирается пороговое число задач, агрегатор вызывается, не дожидаясь grace period. Grace period — минимальное ожидание хвоста, MaxDelay — максимальное; порог размера перекрывает оба ожидания. ### Надёжность: ретраи, тесты, наблюдаемость Рассылка прощает задержки, но не потерю письма, и каждый урок курса здесь превращается в строку конфига или тест. Ретраи: SMTP деградирует редко, но метко — дефолтный экспоненциальный backoff из шестого урока растянет пять попыток на часы, а кастомный RetryDelayFunc из тринадцатого умеет уважать Retry-After почтового провайдера. Тесты: хендлеры — обычные функции, поэтому подход двадцать второго урока ложится без адаптации — fake-хранилище на карте, fake-mailer, который пишет в срез, и проверки на трёх сценариях: подписан, отписался, битый payload. Middleware из двадцать третьего урока пишет тип, ID, очередь и длительность каждой задачи — эти строки и метрики отвечают на вопрос «почему письмо ушло дважды» быстрее любой догадки. Дашборд asynqmon из шестнадцатого урока ставится рядом одной командой и показывает scheduled-окна, группы и латентность очередей вживую. Graceful shutdown из девятнадцатого закрывает деплой: SIGTSTP останавливает забор новых задач, SIGTERM даёт активным письмам дописаться. Восстановитель из глубин курса довершает картину: воркер, умерший посреди письма, отдаст задачу обратно, IsOrphaned подскажет такие случаи в списке активных. Ошибки хендлеров классифицированы, payload валидируется до отправки — грабли двадцать четвёртого урока сняты заранее. > **Про песочницу: письмо без SMTP.** Песочница сайта исполняет программу-имитацию на чистом Go: fake-mailer печатает письма в stdout, очередь эмулируется срезом с задержками — смысл постановки, отмены и агрегации виден без Redis и без Docker. Локально всё по-настоящему: Go, Redis в контейнере, а рядом — asynqmon для живого дашборда: scheduled-окна welcome, группы дайджеста и латентность очередей видны в тот же момент, когда вы жмёте кнопку подписки. ### Чек-лист перед запуском - Постановка welcome идемпотентна: детерминированный TaskID из email, дубль ловится ошибкой. - У welcome есть окно отмены: ProcessIn, удаление из очереди при отписке, проверка в хендлере. - Дайджест собирается группами: GracePeriod, MaxDelay и MaxSize подобраны под трафик. - Очереди разведены: дайджест в critical, одиночные письма в default, веса проверены под нагрузкой. - Ошибки классифицированы: битые payload — SkipRetry, SMTP и хранилище — ретрай с backoff. - Хендлеры протестированы без Redis: подписан, отписался, битый payload. - Middleware пишет тип, ID, очередь и длительность; метрики уходят в мониторинг. - Деплой переживёт SIGTERM: graceful shutdown включён и отрепетирован. - Периодика дайджеста описана cron-выражением и проверена на локальном расписании. ### Соберите постановку welcome Допишите опции: отложка на пять минут и детерминированный идентификатор задачи. ```go _, err = s.client.Enqueue(asynq.NewTask(TypeWelcomeEmail, payload), asynq.___(5*time.Minute), asynq.___("welcome:"+email), ) ``` Подсказка: Первая опция принимает Duration, вторая — строку с идентификатором. Правильные ответы в пропусках: `ProcessIn`, `TaskID` ### Соберите конвейер подписки Расставьте шаги хендлера подписки: от декода тела до ответа клиенту. Правильный порядок строк: 1. `var req SubscribeRequest` 2. `if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "битое тело", 400); return }` 3. `email := strings.ToLower(strings.TrimSpace(req.Email))` 4. `if err := s.store.CreateSubscriber(r.Context(), email); err != nil { http.Error(w, "уже подписан", 409); return }` 5. `payload, _ := json.Marshal(WelcomePayload{Email: email})` 6. `task := asynq.NewTask(TypeWelcomeEmail, payload)` 7. `s.client.Enqueue(task, asynq.ProcessIn(5*time.Minute),` 8. ` asynq.TaskID("welcome:"+email), asynq.Queue("default"))` 9. `w.WriteHeader(http.StatusAccepted)` ### Карточки: итоговый проект - **ProcessIn(5 * time.Minute)** — Опция постановки: задача уходит в scheduled и исполнится через пять минут. Окно, за которое успевает прийти отписка. - **TaskID("welcome:" + email)** — Детерминированный ID задачи: повторная постановка на ту же почту вернёт ошибку дубликата — идемпотентность на входе конвейера. - **GroupAggregatorFunc** — Функция-агрегатор: получает группу и задачи, возвращает одну общую. Вместе с GroupGracePeriod, GroupMaxDelay и GroupMaxSize задаёт пачку дайджеста. - **Два слоя отмены welcome** — DeleteTask через Inspector для scheduled-задачи плюс перечитывание подписки в хендлере перед отправкой. Каждая защита закрывает свою гонку. - **Периодика дайджеста** — Cron-постановка с ProcessAt на следующий слот или PeriodicTaskManager с провайдером PeriodicTaskConfig — когда конфиг меняется без деплоя. - **Тесты без Redis** — Хендлеры — чистые функции: fake-хранилище и fake-mailer закрывают сценарии подписан, отписался, битый payload. Проект собран, и чек-лист курса закрыт целиком: постановка из веба, отложки и окно отмены, идемпотентные ID, очереди с приоритетами, группы и агрегация, ретраи с backoff, тесты без Redis, middleware, дашборд и graceful shutdown — двадцать четыре урока превратились в один работающий сервис. Мини-эксперимент: поднимите Redis, поставьте welcome через ProcessIn на пять минут, откройте asynqmon, найдите задачу в scheduled, отставьте её RunTask'ом — и сравните поведение с отпиской, которая удаляет письмо DeleteTask'ом. Дальше три дороги. Первая — [курс по Redis](/courses/go-redis/): очередь asynq — тонкий слой над структурами данных Redis, и понимание списков, отсортированных множеств и pubsub объяснит, куда уходит память под тысячи scheduled-задач. Вторая — [курс по cron](/courses/cron/): cron-выражения из девятого урока станут родным языком. Третья — регулярная практика: возьмите каркас этого проекта, добавьте подтверждение почты по токену, отписочную ссылку с одноразовым ключом и шаблоны писем — и у вас будет продакшн-рассылка, а не учебная. ## Коротко: что запомнить - Итоговый проект — система email-рассылок: два типа задач, две очереди, HTTP-сервис плюс воркеры. - Welcome ставится из хендлера с ProcessIn(5 * time.Minute) — окно на отмену для отписки. - TaskID("welcome:" + email) делает постановку идемпотентной: дубль ловится ошибкой дубликата. - Отмена письма работает в два слоя: DeleteTask из очереди плюс проверка подписки в хендлере. - Дайджест — периодическая задача; уведомления собираются группами через GroupAggregatorFunc. - GracePeriod, MaxDelay и MaxSize управляют терпением группы; MaxSize собирает пачку немедленно. - Ретраи с backoff спасают от деградации SMTP, SkipRetry отсекает битые payload. - Хендлеры тестируются без Redis, деплой закрывает graceful shutdown, asynqmon показывает всё вживую. ## Домашнее задание Соберите проект в трёх пакетах — api, tasks, mailer — и запустите целиком: HTTP-хендлер подписки с идемпотентной постановкой welcome, хендлер письма с проверкой подписки, PeriodicTaskManager или cron-постановку дайджеста и агрегатор групп. Снимите три сценария тестами без Redis: повторная подписка не создаёт дубль, отписка в окне пяти минут оставляет письмо неотправленным, битый payload не должен ретраиться. Поднимите asynqmon и посмотрите, как welcome живёт в scheduled и исчезает при отписке — этот сервис станет вашим стартовым шаблоном. ## Вопросы и ответы по уроку ### Почему окно отмены именно пять минут? Пять минут — компромисс из проекта: человек успевает передумать, а бизнес не копит ненужных ожиданий. Значение задаётся ProcessIn в одном месте и проверяется нагрузкой: чем длиннее окно, тем больше задач копится в scheduled и тем дольше жить записи о подписке до подтверждения. ### Почему проверка подписки в хендлере, если задача уже удалена при отписке? DeleteTask не трогает исполняемые задачи, а гонка «письмо уже в active» вполне реальна. Хендлер читает актуальное состояние непосредственно перед отправкой — это второй слой, который гарантирует: неактуальное письмо не уйдёт, даже если отменить из очереди не удалось. ### Что брать для периодики — cron-постановку или PeriodicTaskManager? Cron-постановка проще и достаточно, когда расписание зашито в код и не меняется. PeriodicTaskManager берут, когда конфигурация приезжает из хранилища или конфига и должна применяться без перезапуска: провайдер опрашивается периодически, и новые PeriodicTaskConfig подхватываются живым менеджером.