LISTEN/NOTIFY: события из базы

Кэш пользователей инвалидируется соседним сервисом, воркеру нужно проснуться, когда в таблице появилась задача, вебсокет-шлюзу — разослать клиентам смену статуса заказа. Типовое решение — опрос: каждые пять секунд SELECT по всей ферме приложений. Десять подов, два опроса в минуту каждый — двадцать холостых запросов в минуту ради событий, которые случаются раз в час; задержка обнаружения в среднем равна половине интервала, а при удачном стечении пауз события и вовсе копятся пачками. Postgres решает это наоборот: NOTIFY публикует событие на канал, LISTEN подписывает сессию, и уведомление доставляется всем слушателям мгновенно. Обратная сторона — правила, на которых ломаются первые прототипы: уведомления не хранятся, не ретраятся и умирают вместе с соединением слушателя. Урок идёт по порядку: механика каналов и структура pgconn.Notification, атомарность с транзакцией, дизайн payload и лимиты очереди, переподключение, правило выделенного коннекта и сравнение с альтернативами.

Механика: каналы, сессии и pgconn.Notification#

Подписка — обычная SQL-команда: conn.Exec(ctx, "LISTEN events") привязывает слушание канала events к конкретной сессии. Канал создаётся фактом упоминания — никакой регистрации нет; UNLISTEN events снимает одну подписку, UNLISTEN — все разом, а закрытие сессии убирает их автоматически. Публикация — команда NOTIFY events, 'текст' или функциональная форма pg_notify('events', 'текст'), которую можно вызвать из любого запроса и из триггера. Канал — идентификатор, а не строка: без кавычек он приводится к нижнему регистру, EVENTS и events — один канал, в кавычках регистр сохраняется; рабочая конвенция — нижний регистр без кавычек и ноль сюрпризов. На стороне Go получение выглядит так: ntf, err := conn.WaitForNotification(ctx) — вызов блокируется до события или отмены контекста и возвращает pgconn.Notification с тремя полями: Channel — имя канала, Payload — строка, отправленная вместе с событием, PID — идентификатор серверного процесса-отправителя, по которому удобно фильтровать собственные события. Приятная деталь из исходников pgx: уведомления, приехавшие посреди других запросов на этом же коннекте, драйвер складывает в буфер и отдаёт при следующем WaitForNotification — по дороге ничего не теряется. Каналов на одну сессию можно подписать сколько угодно: каждая команда LISTEN добавляет подписку, один цикл WaitForNotification обслуживает их все, а внутри слушателя события раскладываются по Channel — так один выделенный коннект кормит и инвалидацию кэша, и будильник воркера сразу.

Слушающий коннект — выделенный: WaitForNotification занимает соединение на неопределённый срок, конкурентные вызовы на одном *pgx.Conn невозможны — вторая операция упадёт с conn busy, та же механика занятого коннекта, что в пятнадцатом уроке. Поэтому обычные запросы гоняют через пул из четвёртого урока, а слушатель живёт на отдельном pgx.Connect или на выделенном из пула коннекте, который не возвращают. Самоуведомление тоже стоит знать заранее: сессия, которая и слушает, и публикует, получит собственные события — фильтруйте их по PID или кладите в payload признак источника. И ещё одно: уведомление видят только сессии, уже выполнившие LISTEN к моменту доставки — подписаться задним числом нельзя, история канала не ведётся. Три свойства — блокировка, самоуведомление, отсутствие истории — объясняют почти каждую граблю дальше по тексту. Зато API остаётся крошечным: одна команда подписки, один блокирующий вызов, одна структура — вся сложность живёт в правилах эксплуатации, а не в сигнатурах.

Слушатель с переподключением

func listenEvents(ctx context.Context, dsn string) error {
        for {
                err := listenOnce(ctx, dsn)
                if ctx.Err() != nil {
                        return ctx.Err()
                }
                if err != nil {
                        log.Printf("слушатель упал: %v, retry через 2s", err)
                        time.Sleep(2 * time.Second)
                }
        }
}

func listenOnce(ctx context.Context, dsn string) error {
        conn, err := pgx.Connect(ctx, dsn)
        if err != nil {
                return err
        }
        defer conn.Close(ctx)

        if _, err = conn.Exec(ctx, "LISTEN events"); err != nil {
                return err
        }
        for {
                ntf, err := conn.WaitForNotification(ctx)
                if err != nil {
                        return err
                }
                log.Printf("pid=%d канал=%s payload=%s",
                        ntf.PID, ntf.Channel, ntf.Payload)
        }
}

Внешний цикл существует потому, что подписка умирает вместе с сессией: после любого сбоя listenOnce ставит LISTEN заново — переподключение и переподписка неразделимы. WaitForNotification блокируется до события или отмены контекста: дедлайн из пятнадцатого урока — штатный способ остановить слушателя при завершении сервиса. Ошибка из внутреннего цикла означает не «кончились события», а «сессия мертва», и она обязана возвращаться наружу — иначе слушатель замирает молча.

Атомарность: NOTIFY и транзакция#

NOTIFY внутри транзакции не улетает сразу: сервер копит уведомления и рассылает их только при COMMIT, в порядке выполнения команд. ROLLBACK уничтожает всё, что успел NOTIFY, — и это фича, а не подлость: подписчик никогда не узнает о строке, которой нет, уведомление и данные атомарны, ровно как INSERT и UPDATE из тринадцатого урока. Десять NOTIFY в одной транзакции — десять событий после одного коммита, и дедупликации нет: два одинаковых NOTIFY дадут два уведомления, слушатель обязан переживать повторы. Команда NOTIFY вне явной транзакции — это автокоммит: уведомление уходит сразу после выполнения, что удобно для ручных тестов из psql. Вторая половина атомарности — доставка: она мгновенная и одноразовая. Если в момент COMMIT слушающих сессий нет, событие уходит в пустоту и не вернётся никогда — ни истории, ни повторного прогона. Отсюда архитектурное правило, которое экономит ночи: NOTIFY — это звонок «посмотри в таблицу», а не сама посылка; данные живут в таблице, уведомление — только сигнал, а потерянный сигнал добирается периодическим опросом той же таблицы как страховкой.

Живое место: Release, который молча убил подписку

Слушателя выделяют из пула: acquired, err := pool.Acquire(ctx), затем conn := acquired.Conn() и LISTEN. Грабля в руке, которая по привычке из каждого примера с пулом пишет defer acquired.Release() сразу после настройки. Если Release выполнится раньше конца жизни слушателя, то коннект вернётся в пул: его заберёт обычный запрос, уведомления начнут капать в чужую сессию, а по MaxConnLifetime пул закроет соединение — и подписка исчезнет без единой ошибки в логах. Если Release обязан быть, то он ставится на путь graceful shutdown, а не сразу после LISTEN; для долгоживущего слушателя честнее отдельный pgx.Connect — чужие коннекты не занимать.

Проверка: момент доставки

Издатель выполнил NOTIFY внутри транзакции, транзакция ещё не завершена. Что увидят слушатели?

Payload и очередь: лимиты, о которых забывают#

Payload NOTIFY ограничен 8000 байтами — превышение ломает команду на отправке, но потолок не повод втискивать туда целую строку. Рабочий дизайн — JSON из идентификаторов: id заказа, тип события, имя источника; слушатель дочитывает данные из таблицы и всегда видит актуальное состояние, даже если строка изменилась между NOTIFY и чтением. Дедупликации нет: два одинаковых NOTIFY — два уведомления, и «разослал всем по кругу» легко превращается в лавину повторов. Нагрузку видит очередь уведомлений — она одна на всю базу: если издатели пишут быстрее, чем слушатели вычитывают, очередь растёт, и pg_notification_queue_usage() показывает занятую долю от нуля до единицы. Алерт на пороге 0.5 — дешёвый монитор мёртвого слушателя: живой вычитывает очередь быстрее, чем она наполняется. Payload — просто текст в кодировке базы: никакой структуры протокол не навязывает, но JSON с полями разбирать легче, чем самодельные разделители, которые ломаются на первом же переносе строки. При полном переполнении транзакции с NOTIFY падают на коммите — база останавливает издателей, чтобы очередь не съела память; до этого лучше не доводить, но механику знать стоит.

Издатель: pg_notify внутри транзакции

func publishOrder(ctx context.Context, pool *pgxpool.Pool,
        o Order) error {

        return pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
                _, err := tx.Exec(ctx,
                        "INSERT INTO orders (user_id, total) VALUES ($1, $2)",
                        o.UserID, o.Total)
                if err != nil {
                        return err
                }
                payload := fmt.Sprintf("{\"order_id\":%d,\"user_id\":%d}",
                        o.ID, o.UserID)
                _, err = tx.Exec(ctx, "SELECT pg_notify($1, $2)",
                        "events", payload)
                return err
        })
}

pg_notify — функциональная форма NOTIFY, и она же живёт в триггере AFTER INSERT: продакшен-паттерн, при котором база отправляет уведомление сама, а приложение физически не может забыть. Payload несёт только идентификаторы — слушатель дочитает заказ из таблицы, 8000 байт не потревожатся. Обёртка BeginFunc из четырнадцатого урока коммитит в конце: лишь после этого событие уйдёт слушателям, атомарно с INSERT.

Что получит слушатель?

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

return pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
    _, err := tx.Exec(ctx,
        "NOTIFY events, 'order-created'")
    if err != nil {
        return err
    }
    return errors.New("валидация не прошла")
})
// defer tx.Rollback() сработает
// слушатель висит на WaitForNotification

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

Сервер хранит подписки в памяти сессии: соединение умерло — LISTEN исчез, никакой «восстанови мою подписку» не существует. WaitForNotification на мёртвом коннекте вернёт ошибку — это не повод выходить из программы, а сигнал начать цикл переподключения: pgx.Connect заново, LISTEN заново, продолжить читать. Между падением и повторным LISTEN — слепое пятно: события этого окна потеряны безвозвратно, и окно растёт на паузах бэкоффа. Поэтому продакшен-слушатель держит данные в таблице, а уведомление трактует как «проснись и проверь»: пропуски закрывает периодический фоновый опрос, дешёвый раз в минуту, а уведомление лишь ускоряет реакцию с минуты до миллисекунд. Второй типовой приём — pg_notify из триггера AFTER INSERT: база шлёт событие при вставке строки, издатель не может забыть его отправить, payload несёт идентификатор. Связка «триггер публикует, слушатель вычитывает, фоновый опрос страхует» закрывает и переподключения, и слепые окна — минимальный каркас, с которого стоит копировать.

Живое место: событие, отправленное в пустоту

Деплой поднял издателя на минуту раньше слушателя: миграции, старт, первые заказы — и три NOTIFY ушли в базу, где ещё никто не сделал LISTEN. Если слушатель стартует позже издателя, то события стартового окна не вернутся никогда: доставка одноразовая, истории нет. Лекарство двухслойное: readiness-проба слушателя переводит сервис в готовность только после успешного LISTEN, а события с бизнес-требованиями живут в таблице-журнале — NOTIFY лишь сокращает задержку и не является гарантией доставки.

NOTIFY против опроса и очередей#

Опрос — постоянная нагрузка и задержка до половины интервала, зато нулевая магия: рестарты, деплои и мёртвые коннекты ему безразличны, источник истины всегда в таблице. NOTIFY — миллисекунды и ноль холостых запросов, но гарантия at-most-once: без ретраев, без персистентности, без порядка между разными сессиями. Очереди — Redis Streams, Kafka — дают персистентность, ретраи, группы потребителей и порядок в партиции, но это отдельная система со своей эксплуатацией. Выбор делается по роли события: инвалидация кэша прощает потерю — ключ всё равно протухнет по TTL, и NOTIFY здесь идеален; «проснись, есть работа» работает в связке с таблицей задач, где потерянный сигнал добирает фоновый скан; платежи, заказы и интеграции требуют гарантий — им дорога в настоящую очередь, например из курса по go-redis, а не в NOTIFY. Правило здравого смысла: NOTIFY не заменяет очередь — он убирает опрос там, где потеря события стоит дешевле постоянного SELECT. Гибрид тоже легитимен: очередь из таблицы, NOTIFY как ускоритель — базовая механика остаётся надёжной, задержка падает на порядки.

Три способа узнать о событии#

Механизм	Задержка	Гарантии	Рестарт слушателя	Типичная роль
Опрос таблицы	До половины интервала	Определяются данными в таблице	Переживает	Фоновые джобы без реалтайма
LISTEN/NOTIFY	Миллисекунды	At-most-once: без ретраев и истории	Не переживает: подписка умирает	Инвалидация кэша, будильник воркера
Redis Streams / Kafka	Миллисекунды	Персистентность, ретраи, группы	Переживает	Заказы, платежи, интеграции

Правила LISTEN/NOTIFY одним списком#

Слушатель — на выделенном коннекте: pgx.Connect или Acquire без Release
LISTEN ставится заново после каждого переподключения: подписка умирает с сессией
Уведомление уходит только при COMMIT; ROLLBACK стирает его вместе с данными
Пропущенное событие не вернуть: NOTIFY — сигнал, данные — в таблице
Payload до 8000 байт: идентификаторы в JSON, строку дочитывают из базы
Дедупликации нет: слушатель обязан переживать повторы
pg_notification_queue_usage() — монитор очереди; алерт на пороге 0.5
Канал — идентификатор: нижний регистр без кавычек, никаких сюрпризов

Проверка: выделенный коннект

Почему слушателя держат на отдельном pgx.Connect, а не на обычном коннекте из пула?

Соберите регистрацию слушателя

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

conn, err := pgx.(ctx, dsn) // отдельная сессия, не пул if err != nil { return err } defer conn.(ctx) if _, err = conn.Exec(ctx, " events"); err != nil { return err } for { ntf, err := conn.(ctx) if err != nil { return err } log.Printf("%s -> %s", ntf.Channel, ntf.Payload) }

Соберите тело listenOnce

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

Строки

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

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

Карточки: LISTEN/NOTIFY

Термин

pgconn.Notification

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

Определение

Структура с тремя полями: Channel, Payload и PID — идентификатор серверного процесса-отправителя, по нему фильтруют собственные события.

1 / 6

Песочница урока не исполняет pgx: LISTEN/NOTIFY требует живого Postgres, поэтому слушатель и издатель гоняются локально — поднимите контейнер из второго урока, откройте два терминала: в одном слушатель, во втором psql с командой NOTIFY events, 'test'. Дальше по курсу: тринадцатый урок — каркас транзакций, на котором держится атомарность уведомлений; четвёртый — пул и владение коннектами, из которых вырастает правило выделенного слушателя; пятнадцатый — отмена контекста, которой слушатель корректно останавливается. Домашнее задание собирает всё в один рабочий слушатель с переподключением — навык, который на собеседовании отличает «знаю про NOTIFY» от «эксплуатировал». Одна оговорка напоследок: LISTEN/NOTIFY — серверная механика Postgres, а не pgx, поэтому всё изученное здесь переносится на любой драйвер и язык — pgx просто даёт самый удобный доступ к ней в Go.

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

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

Вывод

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

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

LISTEN подписывает сессию на канал; NOTIFY и pg_notify рассылают события слушателям за миллисекунды.
WaitForNotification блокируется до события или отмены контекста и возвращает pgconn.Notification: Channel, Payload, PID.
Уведомление уходит только при COMMIT; ROLLBACK стирает его вместе с данными транзакции.
Персистентности нет: событие без слушателя пропадает — NOTIFY сигнал, источник истины в таблице.
Payload до 8000 байт, дедупликации нет, очередь общая на базу: pg_notification_queue_usage() показывает заполнение.
Слушатель живёт на выделенном коннекте: отдельный pgx.Connect или Acquire без раннего Release.
Сессия умерла — LISTEN умер: цикл переподключения ставит подписку заново, слепое окно страхует опрос.
NOTIFY идеален для инвалидации кэша и будильника воркера; гарантии доставки — работа настоящих очередей.

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

Поднимите Postgres и соберите слушателя из урока: отдельный pgx.Connect, LISTEN events, цикл WaitForNotification с логом канала, payload и PID. Вторым процессом публикуйте события из psql командой NOTIFY events, 'test' и добейтесь доставки за секунду. Затем сломайте слушателя — остановите процесс или рестартуйте базу — и убедитесь, что WaitForNotification вернул ошибку, а цикл переподключился и выполнил LISTEN заново: без повторной подписки событий больше не будет, проверьте это. Отправьте NOTIFY внутри транзакции и откатите её — слушатель должен промолчать. В конце выполните pg_notification_queue_usage() до и после сотни NOTIFY без слушателя и запишите в комментарий, какой порог алерта вы бы поставили и почему.

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

Можно ли слушать канал на коннекте из пула и продолжать обычные запросы?

Подписка ставится любым Exec, но использовать этот коннект для запросов нельзя: WaitForNotification занимает его надолго, а конкурентные вызовы на одном соединении падают с conn busy. Драйвер, к слову, буферизует уведомления, приехавшие во время других запросов на том же коннекте, — но практический паттерн один: выделенный коннект только для слушания, обычная работа через пул. Отдельный pgx.Connect на слушателя стоит одно соединение из лимита — планируйте MaxConns с этим запасом.

Получит ли сессия собственное уведомление, если она сама слушает канал?

Да, самоуведомление работает: сессия-издатель получит своё событие, если она же выполнила LISTEN. Фильтруйте такие события по PID отправителя или по признаку источника в payload — иначе воркер обработает задачу, которую сам же поставил.

Чем NOTIFY хуже настоящей очереди и когда это приемлемо?

У уведомлений нет персистентности, ретраев, подтверждений и порядка между сессиями: это at-most-once сигнал. Приемлемо там, где потеря события дешева: инвалидация кэша, будильник воркера, обновление счётчиков на дашборде. Для платежей и интеграций нужна очередь с гарантиями — таблица задач с фоновым опросом или внешний брокер.

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

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