---
title: "Inspector: мониторинг из кода"
description: "Asynq Inspector: мониторинг очереди из кода — GetQueueInfo и Latency, списки задач с TaskInfo, RunTask, ArchiveTask, DeleteTask, CancelProcessing. Примеры кода."
course: "Asynq в Go: очередь фоновых задач на Redis"
url: "https://golangmaster.ru/courses/asynq/lessons/15-inspektor/"
published: "2026-09-19"
updated: "2026-09-23"
minutes: 20
---

# Inspector: мониторинг из кода

*Курс «Asynq в Go: очередь фоновых задач на Redis», урок 15 из 25. ~20 минут. Обновлено 23.09.2026. Источник: https://golangmaster.ru/courses/asynq/lessons/15-inspektor/*

Тикет «письма не уходят», а терминал пуст — нужен доступ к очереди из кода, и asynq Inspector даёт мониторинг очереди программно: статистика, списки задач, ручной запуск и отмена. Соберём дежурный скрипт и админ-ручку, разберём ошибки и границы операций.

Воркеры живы, Redis жив, а письма стоят: что происходит — не видно ни из логов приложения, ни из терминала. Первый рефлекс — открыть asynqmon из шестнадцатого урока, но дашборд отвечает на вопрос «что происходит», а не «что с этим делать сейчас»: дежурному инженеру нужен скрипт, админке — ручка, алертеру — метрика. Для всего этого у asynq есть Inspector — клиент управления очередью: те же данные, что видит дашборд, но в виде методов Go. Читают они одни и те же ключи Redis, поэтому числа в asynqmon и в вашем скрипте совпадают; разница только в форме доступа. Он умеет показывать статистику очереди и отдельных задач, перечислять задачи по состояниям и вмешиваться: запускать застрявшие, архивировать, удалять, отменять исполняемые. В этом уроке соберём мониторинг и вмешательство из кода: коннект, статистика, списки, ручные операции и их границы — с акцентом на то, где Inspector безопасен, а где легко сломать живую очередь.

### Asynq Inspector: коннект и статистика

Asynq Inspector создаётся тем же коннект-оптом, что клиент и сервер: inspector := asynq.NewInspector(asynq.RedisClientOpt{Addr: "localhost:6379"}) — и работает независимо от воркеров: его можно встроить в админ-сервис, CLI-утилиту или тест. Базовая операция — GetQueueInfo(queue): метод возвращает QueueInfo со срезом очереди целиком — счётчики Pending, Active, Scheduled, Retry, Archived, Completed и Aggregating, суммарный Size, память, которую очередь занимает в Redis, и счётчики обработок: Processed и Failed за текущие сутки плюс накопительные ProcessedTotal и FailedTotal. Отдельного слова заслуживает поле Latency — возраст самой старой задачи в pending: это самый честный сигнал затыка, потому что растёт он именно тогда, когда очередь не успевает, а не когда задач много. Поле Paused показывает, стоит ли очередь на паузе — да, очередью тоже можно управлять, и об этом ниже. Поле MemoryUsage честно помечено в документации как приближение: значение получается сэмплированием, так что сравнивайте очереди между собой, а не байты с бухгалтерией. Рядом лежит Timestamp — момент снятия среза: в логах метрик он спасает, когда скрипт отстал от реальности. Свежая операция — UpdateTaskPayload из версии 0.26: метод меняет payload задачи, пока она сидит в scheduled; активную или ожидающую в pending он не тронет и вернёт ошибку. Список имён всех очередей даёт метод Queues() — с него удобно начинать обход в скрипте мониторинга.

### Срез очереди из кода

```go
inspector := asynq.NewInspector(asynq.RedisClientOpt{Addr: "localhost:6379"})
defer inspector.Close()

info, err := inspector.GetQueueInfo("default")
if err != nil {
    log.Fatal(err)
}
log.Printf("pending=%d active=%d retry=%d archived=%d latency=%v",
    info.Pending, info.Active, info.Retry, info.Archived, info.Latency)
```

Inspector строится на RedisClientOpt, как клиент: отдельного порта и регистрации не нужно, а Close освобождает соединение. GetQueueInfo возвращает срез очереди одним вызовом; Latency — возраст самой старой pending-задачи, и именно по нему строят алерт на затык очереди. Логируйте срез целиком, а не только Latency: пара «растущий Archived при плоском Processed» расскажет о фатальных ошибках раньше тикетов.

### Мониторинг очереди: Latency и история

Мониторинг очереди живёт на трёх числах. Первое — Latency из QueueInfo: возраст самой старой задачи в ожидании; порог задаётся бизнесом — для писем минуты, для платежей секунды. Второе — история: метод History(queue, n) возвращает DailyStats за последние n дней — Processed и Failed по суткам; на этих числах строятся графики пропускной способности и сравнение «вчерашний пик против сегодняшнего». DailyStats несёт ровно четыре поля: имя очереди, Processed, Failed и дата — достаточно для графика, ничего лишнего для путаницы. Обратите внимание: Processed включает и успешные, и упавшие задачи, — долю Failed считайте в своём скрипте, готовой её никто не отдаёт. Третье — соотношения внутри среза: растущий Retry при плоском Processed означает, что зависимый сервис деградирует; растущий Archived — что фатальные ошибки из четырнадцатого урока пошли потоком; большое Aggregating — что группы из десятого урока перестали собираться. Пороговые значения вычисляются у вас, а не в библиотеке: Inspector честно отдаёт сырые числа, а «много» для писем и «много» для платежей — разные вещи. Практичная связка: cron-задача из девятого урока каждые пять минут вызывает GetQueueInfo и шлёт метрики в вашу систему наблюдаемости — тем самым очередь становится частью обычного дашборда, а не отдельным миром, в который заглядывают только при инциденте.

### Проверка: что такое Latency

**Вопрос:** Поле Latency в QueueInfo показывает...

- Среднее время исполнения задач за сутки
- ✅ Возраст самой старой задачи в состоянии pending
- Задержку сети между воркером и Redis
- Суммарное время задач в retry-состоянии

Latency измеряется возрастом самой старой задачи, ждущей в pending: если очередь не успевает, это число растёт. Среднее время исполнения сюда не входит, а сетевые задержки — забота метрик Redis, а не Inspector.

### Списки задач: List-методы и TaskInfo

За срезом очереди стоят списки задач: ListPendingTasks, ListActiveTasks, ListScheduledTasks, ListRetryTasks, ListArchivedTasks, ListCompletedTasks и ListAggregatingTasks — по одному на состояние. Все методы принимают имя очереди и опции пагинации: asynq.PageSize(n) и asynq.Page(n), где страницы считаются с единицы; без опций вернутся первые тридцать задач — этого хватает для взгляда, но не для обхода, поэтому в скриптах пагинацию листают до конца. Для aggregating-состояния метод отдельный и принимает ещё имя группы: ListAggregatingTasks(queue, group) показывает задачи конкретной пачки из десятого урока. Рядом со списками задач есть и SchedulerEntries — зарегистрированные периодические задачи из девятого урока с их cron-выражениями и временем следующего запуска; для диагностики «а запланировано ли вообще что-то» это первый метод, куда смотреть. Каждый элемент — TaskInfo с полным досье: ID, Type, Payload, состояние, MaxRetry и Retried, LastErr с текстом последней ошибки, LastFailedAt, NextProcessAt для отложенных и ретраев, Timeout и Deadline, поле IsOrphaned для задач, чей воркер умер посреди исполнения. Одиночная задача по ID — GetTaskInfo(queue, id): дежурный сценарий «пользователь дал ID из письма — скажите, что с задачей» реализуется в одну строку. Errors тоже продуманы: несуществующая очередь вернёт ошибку с ErrQueueNotFound, отсутствующая задача — с ErrTaskNotFound, и обе проверяются errors.Is — это не «случайные» ошибки, а часть контракта API.

### Обход retry-задач с ошибками

```go
tasks, err := inspector.ListRetryTasks("default", asynq.PageSize(50))
if err != nil {
    log.Fatal(err)
}
for _, t := range tasks {
    log.Printf("task=%s retried=%d/%d next=%v err=%q",
        t.ID, t.Retried, t.MaxRetry, t.NextProcessAt, t.LastErr)
}
```

ListRetryTasks с PageSize(50) отдаёт до пятидесяти задач retry-состояния; для полного обхода листайте страницы через asynq.Page. Страницы считаются с единицы: Page(1) — первая порция, а нулевая просто трактуется как первая, и вы получите начало списка. Каждый TaskInfo несёт Retried против MaxRetry, время следующей попытки и текст последней ошибки — тройка, по которой ставится диагноз «что и почему ретраится».

> **Живое место: тридцать задач и 500-я ручка.** Два грабли из практики. Первый: List без PageSize тянет первые тридцать задач, и админка честно показывает «задач нет» там, где их тысячи — пагинацию листайте до конца, иначе мониторинг будет показывать только верхушку. Второй: ручка GetTaskInfo без errors.Is(ErrTaskNotFound) отдаёт 500 на каждую задачу, которую очередь уже выполнила и удалила, — а таких в админ-запросах большинство. Проверяйте сентинеллы и возвращайте осмысленное «задача не найдена, возможно, уже выполнена».

### Ручные операции: RunTask, ArchiveTask, DeleteTask

Вмешательство — вторая половина API, и у каждой операции есть чёткие границы по состояниям. RunTask(queue, id) переводит задачу в pending немедленно: работает для scheduled, retry и archived — так запускают вручную отложенное, ретраи и задачи из архива после починки данных; для pending и active метод вернёт ошибку — задача и так в работе. ArchiveTask(queue, id) отправляет в архив задачи из pending, scheduled и retry — мягкая остановка для постановок, которые стали не нужны; активную задачу так остановить нельзя. DeleteTask(queue, id) удаляет задачу совсем: pending, scheduled, retry и archived — но не active. Общая логика границ проста: активную задачу нельзя ни удалить, ни архивировать, потому что воркер уже держит её в руках — для неё есть только отмена исполнения, о которой ниже. Ошибки те же: ErrTaskNotFound для уже исчезнувших задач — в ручных операциях это нормальный исход при гонке с выполнением. Типичный цикл обхода в скрипте: Page(i) в цикле, пока возвращаемая страница полна, — как только она короче PageSize, вы дошли до конца состояния. Для массовых операций есть All-варианты: RunAllArchivedTasks, ArchiveAllScheduledTasks, DeleteAllRetryTasks и другие, знакомые по десятому уроку групповым RunAllAggregatingTasks — они возвращают число обработанных задач и так же уважают активные; логируйте это число, «архивировано 1240 задач» честнее молчаливой массовой операции. Отдельно стоит DeleteQueue(queue, force): с force=false удаляет пустую очередь, с force=true — даже непустую, лишь бы в ней не было активных задач; пригодится при выводе очереди из эксплуатации.

### Операции Inspector и допустимые состояния

| Операция | Работает для | Откажет для |
| --- | --- | --- |
| RunTask(queue, id) | scheduled, retry, archived | pending, active |
| ArchiveTask(queue, id) | pending, scheduled, retry | active, уже архив |
| DeleteTask(queue, id) | pending, scheduled, retry, archived | active |
| CancelProcessing(id) | исполняемая задача | всех остальных: сигнал просто никого не найдёт |

### Правила ручных операций

- Активную задачу нельзя удалить или заархивировать — только отменить исполнение.
- ErrTaskNotFound при ручной операции — часто гонка с выполнением: трактуйте спокойно.
- Проверяйте State из TaskInfo перед операцией: API не прощает догадок.
- Логируйте, кто и что запустил вручную: ручной RunTask — изменение данных, а не чтение.
- Массовые All-операции стартуют сотни задач разом: считайте нагрузку на зависимые сервисы.
- Операции не идемпотентны: повторный ArchiveTask вернёт ошибку, а не тишину.

### Проверка: запуск активной задачи

**Вопрос:** Что вернёт inspector.RunTask("default", id) для задачи в состоянии active?

- Ничего: метод подождёт завершения и запустит задачу заново
- ✅ Ошибку: RunTask работает только для scheduled, retry и archived
- Задача будет запущена вторым воркером
- Активная задача уйдёт в архив

RunTask переводит задачу в pending, но только если она ждёт: scheduled, retry или archived. Активная задача уже в руках воркера — запуск второй копии сломал бы семантику, поэтому метод возвращает ошибку, а для активных есть CancelProcessing.

### Что сделает отмена с задачей?

```go
// хендлер спит на таймере с select по ctx.Done()
// задача исполняется воркером

err := inspector.CancelProcessing(taskID)
```

**Что выведет программа?**

- Задача удалится без следа
- ✅ Контекст хендлера отменится, задача засчитается неудачей и уйдёт в retry
- Задача мгновенно перезапустится с нуля
- Ошибка: CancelProcessing работает только для pending

CancelProcessing шлёт сигнал отмены, воркер отменяет контекст хендлера, и задача проваливается с ошибкой контекста — дальше работает обычный ретрай по политике из шестого урока. Удалять и перезапускать отмена ничего не делает: это вежливый толчок, а не disposal.

### Соберите срез очереди

Допишите конструктор Inspector и метод статистики очереди.

```go
inspector := asynq.___(asynq.RedisClientOpt{Addr: "localhost:6379"})
info, err := inspector.___("default")
```

Подсказка: Конструктор — NewInspector, статистика — GetQueueInfo с именем очереди.

Правильные ответы в пропусках: `NewInspector`, `GetQueueInfo`

### Соберите дежурный скрипт

Расставьте строки: коннект, срез очереди, обход архива, ручной запуск, закрытие.

Правильный порядок строк:

1. `inspector := asynq.NewInspector(asynq.RedisClientOpt{Addr: "localhost:6379"})`
2. `defer inspector.Close()`
3. `info, err := inspector.GetQueueInfo("default")`
4. `if err != nil { log.Fatal(err) }`
5. `if info.Latency > 10*time.Minute {`
6. `    log.Printf("очередь стоит: latency=%v", info.Latency)`
7. `}`
8. `archived, err := inspector.ListArchivedTasks("default", asynq.PageSize(10))`
9. `for _, t := range archived { log.Printf("архив: %s err=%q", t.ID, t.LastErr) }`

### CancelProcessing: отмена активной задачи

Единственная операция над активными задачами — CancelProcessing(id): она принимает только идентификатор, без очереди, и работает сигналом, а не командой. Внутри — публикация в Redis pubsub-канал, который слушают все ваши серверы: каждый держит реестр исполняемых задач, находит нужный ID и отменяет контекст хендлера у себя. Документация называет метод best-effort, и это важно: результат вызова говорит лишь о том, что сигнал отправлен, — задача могла уже закончиться, сигнал мог прийти на сервер, где её нет, хендлер мог игнорировать контекст. Что происходит с задачей после отмены, определяется общими правилами: контекст отменён — хендлер, слушающий ctx.Done(), возвращает управление, воркер засчитывает неудачу с ошибкой контекста, и задача ретраится по обычной политике; хендлер, игнорирующий контекст, доделает свою работу, как в одиннадцатом уроке. Отсюда честные ожидания: CancelProcessing — это «попросить остановиться», а не «убить»; для повторов, которые должны прекратиться совсем, есть ArchiveTask после отмены или SkipRetry в самом хендлере. И помните: отмена не снимает задачу с очереди — она лишь завершает текущую попытку.

> **Пауза очереди и живые воркеры.** У Inspector есть и выключатель: PauseQueue(queue) останавливает забор задач из очереди, UnpauseQueue возвращает — удобно на деплое и при деградации зависимого сервиса, когда задачи должны подождать, а не копиться в ретраях. Поле Paused в QueueInfo показывает текущее состояние. Метод Servers() вернёт живые воркеры с их Concurrency и списками очередей — быстрый ответ на «а кто вообще обрабатывает эту очередь» без доступа к контейнерам. ServerInfo несёт и подробности: ID, хост, PID, карту очередей с весами, время старта и статус — по нему видно и кто подключён, и с какой конфигурацией.

### Карточки: Inspector

- **asynq.NewInspector** — Конструктор управления: тот же RedisConnOpt, что у клиента и сервера. Живёт независимо от воркеров — в админке, CLI, тестах.
- **GetQueueInfo** — Срез очереди: счётчики по состояниям, Size, память, Processed и Failed за сутки, Latency — возраст самой старой pending-задачи.
- **ListRetryTasks и соседи** — Списки задач по состояниям: Pending, Active, Scheduled, Retry, Archived, Completed, Aggregating. По умолчанию — первые 30, дальше PageSize и Page.
- **TaskInfo** — Досье задачи: ID, Type, Payload, State, MaxRetry, Retried, LastErr, LastFailedAt, NextProcessAt, Timeout, Deadline, IsOrphaned и другие.
- **RunTask** — Немедленный перевод в pending для scheduled, retry и archived. Ручной запуск после починки данных или ускорение отложенного.
- **CancelProcessing** — Best-effort сигнал отмены активной задачи по ID: контекст хендлера отменяется, задача проваливается и ретраится по политике.

Inspector собран в рабочий инструмент: GetQueueInfo для среза и Latency, списки с TaskInfo для диагноза, RunTask и ArchiveTask для вмешательства, CancelProcessing для вежливой остановки активной задачи. Мини-эксперимент: поставьте задачу с ProcessIn(5*time.Minute), найдите её через ListScheduledTasks и запустите вручную через RunTask — она перейдёт в pending и исполнится сразу. Формат прежний: код запускается локально с Go и Redis в Docker, песочница сайта показывает тематическую программу на чистом Go без Redis. Дальше по курсу: [шестнадцатый урок](/courses/asynq/lessons/16-asynqmon-ui/) покажет те же данные глазами — в веб-дашборде asynqmon, а [двадцать третий](/courses/asynq/lessons/23-nablyudaemost/) вплетёт эти числа в логи и метрики; групповые операции для агрегатов уже видели в [десятом уроке](/courses/asynq/lessons/10-grupy-agregaciya/).

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

- Asynq Inspector — клиент управления очередью: тот же RedisConnOpt, независим от воркеров.
- GetQueueInfo отдаёт счётчики по состояниям, Processed и Failed, память и Latency — возраст самой старой pending-задачи.
- List-методы перечисляют задачи по состояниям; по умолчанию 30 задач, пагинация через PageSize и Page.
- TaskInfo — полное досье: Retried против MaxRetry, LastErr, NextProcessAt, IsOrphaned.
- RunTask запускает scheduled, retry и archived; активная задача защищена от удаления и архива.
- CancelProcessing — best-effort сигнал: контекст отменяется, задача ретраится по обычной политике.
- ErrTaskNotFound и ErrQueueNotFound — часть контракта: проверяйте errors.Is.
- PauseQueue ставит очередь на паузу, Servers() показывает живые воркеры и их настройки.

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

Напишите CLI-утилиту с двумя командами: stats печатает GetQueueInfo для всех очередей из Queues(), retry <id> запускает задачу через RunTask с проверкой errors.Is(ErrTaskNotFound). Поставьте задачу с ProcessIn(10*time.Minute), найдите её в ListScheduledTasks и запустите вручную — убедитесь, что NextProcessAt сменился исполнением. Добавьте в stats проверку Latency с порогом и алертным логом, а рядом — ArchiveTask с обязательным флагом подтверждения: массовые операции в админке без подтверждения не живут. Это каркас, который переживёт первый ночной инцидент.

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

### Чем Inspector отличается от asynqmon?

Asynqmon — готовый веб-дашборд: смотрят глазами и руками. Inspector — Go-библиотека: те же данные и операции, но в вашем коде — для скриптов, админ-ручек, алертов и тестов. Дашборд приятно иметь всегда, а Inspector нужен там, где решение должно приниматься программно.

### Можно ли давать Inspector наружу — в админку для дежурных?

Можно, но считайте это административным интерфейсом: RunTask и DeleteTask меняют данные, значит, нужны авторизация, логирование действий и подтверждения на массовых операциях. Только для чтения — GetQueueInfo и списки — безопаснее и закрывает большинство вопросов дежурных без риска.

### CancelProcessing гарантированно останавливает задачу?

Нет: метод best-effort — он лишь шлёт сигнал через pubsub, и гарантии исполнения нет. Хендлер, который слушает ctx.Done(), остановится; игнорирующий контекст — доделает работу. Гарантию даёт только ваш хендлер: проверка контекста между шагами делает отмену реальной.