# Занятие 13: Расширяемость / Асинхронная обработка

> Исходная страница в нативном Markdown: Расширяемость / Асинхронная обработка

---

Индекс LLMS: [llms.txt](/ru/llms.txt)

---

<!-- Generated by scripts/generate_skeleton.py. -->

> [!NOTE]
> Эта страница извлечена из официального PDF в нативный Markdown. Текст слайдов, код и комментарии сохранены в порядке чтения; исходные PDF и HTML остаются эталоном макета.
> **HTML source:** [dev2_13_ext_async.html](https://edu.postgrespro.ru/16/dev2-16/dev2_13_ext_async.html)
> **Archive:** [DEV2-handouts-16.zip](https://edu.postgrespro.ru/16/DEV2-handouts-16.zip)
> **Страниц PDF:** 29

## Page 1: Расширяемость / Асинхронная обработка {#source-page-001}

### Авторские права

© Postgres Professional, 2017–2024 Авторы: Егор Рогов, Павел Лузанов, Илья Баштанов, Игорь Гнатюк Фото: Олег Бартунов (монастырь Пху и пик Бхрикути, Непал)
### Использование материалов курса

Некоммерческое использование материалов курса (презентации, демонстрации) разрешается без ограничений. Коммерческое использование возможно только с письменного разрешения компании Postgres Professional. Запрещается внесение изменений в материалы курса.
### Обратная связь

Отзывы, замечания и предложения направляйте по адресу: edu@postgrespro.ru
### Отказ от ответственности

Компания Postgres Professional не несет никакой ответственности за любые повреждения и убытки, включая потерю дохода, нанесенные прямым или непрямым, специальным или случайным использованием материалов курса. Компания Postgres Professional не предоставляет каких-либо гарантий на материалы курса. Материалы курса предоставляются на основе принципа «как есть» и компания Postgres Professional не обязана предоставлять сопровождение, поддержку, обновления, расширения и изменения.

## Page 2: Темы {#source-page-002}

- Зачем нужна асинхронная обработка данных
- Доступные решения
- Реализация очереди средствами PostgreSQL

## Page 3: Асинхронная обработка {#source-page-003}

**Разнесение во времени возникновения события и его обработки**
- Соображения производительности
  - клиенту не требуется ждать ответа
  - возможность управлять ресурсами для обработки
- Реализация
  - очередь сообщений
  - более сложный вариант: модель публикация — подписка

### Комментарии преподавателя

Идея асинхронной обработки событий состоит в том, что возникновение события и его обработка разносятся во времени.

Например, пользователь хочет получить детализацию расходов на мобильную связь. Детализация формируется несколько минут. Можно показать пользователю «песочные часы» и заставить его ждать (синхронная обработка), а можно выслать детализацию по электронной почте, когда она будет готова (асинхронная обработка).

Другой пример: интеграция двух систем. Первая система обращается ко второй, передавая пакет сообщений. Обработка одного сообщения занимает несколько секунд, но в пакете может оказаться тысяча сообщений. Можно заставить первую систему ожидать получения результата (синхронная обработка), а можно ответить «работаем», обработать сообщения асинхронно и уже затем сообщить результат.

Асинхронная обработка сложнее синхронной, но часто оказывается очень удобной. Она позволяет работать эффективнее (клиенту не надо простаивать, дожидаясь ответа) и управлять ресурсами (обрабатывать события с удобной скоростью и в удобное время, а не немедленно).

(Асинхронная обработка широко применяется и в ядре PostgreSQL. Вспомните режим асинхронной фиксации; процесс контрольной точки; процесс автоочистки.)

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

## Page 4: Внешние системы {#source-page-004}

**RabbitMQ, ActiveMQ, ZeroMQ и т. п.**
- Плюсы
  - эти системы работают
  - следование стандартам (AMQP, JMP, STOMP, MQTT...)
  - гибкость, масштабирование, производительность
- Возможные минусы
  - отдельная система (включающая отдельную СУБД)
  - со своими особенностями настройки, администрирования, мониторинга
  - все сложности построения распределенных систем
  - (отсутствие глобальных транзакций)

### Комментарии преподавателя

Одним из вариантов реализации очередей событий являются внешние системы. Названия многих из них традиционно заканчиваются на MQ — Message Queuing.

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

Но надо понимать, что любая большая система потребует серьезных затрат на ее изучение и внедрение. Потребуется разобраться с особенностями настройки, администрирования, мониторинга. Заметим, что в состав систем работы с очередями входит и отдельная СУБД для надежного хранения очередей.

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

## Page 5: Очередь внутри базы: PgQ {#source-page-005}

**Плюсы**
  - система давно на рынке и широко используется
- Возможные минусы
  - мало гибкости, например, исключительно пакетная обработка
  - плохо документирована
  - внешняя программа-демон

### Комментарии преподавателя

Более простым решением может служить реализация очереди в самой СУБД. Особенно это имеет смысл, если события возникают и обрабатываются на сервере баз данных.

Наиболее известна система PgQ, разработанная в свое время компанией Skype (<https://github.com/pgq)>. Эта система достаточно широко используется и про нее известно, что она работает. Если требуется готовое решение, то ей можно и воспользоваться. Поддерживаются версии PostgreSQL 10+.

Из минусов этого решения отметим:

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

● Отсутствие качественной документации (есть описание API: <https://pgq.github.io/extension/pgq/)>.

● Необходимость во внешней (относительно СУБД) программе, обеспечивающей работу очереди.

## Page 6: Очередь внутри базы: pgmq {#source-page-006}

**Плюсы**
  - система активно развивается
  - легковесная реализация
  - все компоненты внутри базы
- Возможные минусы
  - недостаточная гибкость в управлении на низком уровне
  - слабо документирована
  - только исходные коды, необходимость сборки

### Комментарии преподавателя

Решение заявлено компанией Tembo как легковесная реализация очередей. Расширение pgmq написано на Rust и использует инфраструктуру сборки расширений pgrx.

Очереди и метаинформация представлены таблицами в базе данных, а элементы очереди («сообщения») — записями в таблицах. Сообщения можно помещать в очередь (в том числе по несколько сразу), читать из очереди, удалять и архивировать. Для выполнения этих основных действий (и еще ряда дополнительных) в расширении предусмотрен набор функций. Детали реализации очереди скрыты от пользователя, что упрощает работу.

Из минусов pgmq отметим следующие:

● Недостаточная гибкость в управлении на низком уровне. Связано с тем, что расширение pgmq базируется на фреймворке pgrx, в который встроено, в том числе, управление соединениями с сервером СУБД. Повлиять на него средствами уже собранного pgmq нельзя.

● Документация недостаточно подробна.

● В репозитории отсутствуют готовые пакеты, а сборка расширения может оказаться нетривиальной и ресурсоемкой.

<https://tembo.io/pgmq/>

<https://github.com/tembo-io/pgmq>

## Page 7: Очередь средствами расширения pgmq {#source-page-007}

Создадим базу данных и подключимся к ней:
```sql
=> CREATE DATABASE ext_async;
```
```sql
CREATE DATABASE
```
```sql
=> \c ext_async
```
```text
You are now connected to database "ext_async" as user "student".
```
Расширение pgmq уже собрано и доступно для установки. Выполним команду создания расширения в нашей базе. Все его объекты будут размещены в схеме pgmq:
```sql
=> CREATE EXTENSION pgmq;
```
```sql
CREATE EXTENSION
```
Создадим очередь под названием pgmq\_queue:
```sql
=> SELECT pgmq.create('pgmq_queue');
```
```sql
 create
--------
(1 row)
```
Информация об очередях хранится в таблице meta; посмотреть очереди можно с помощью табличной функции:
```sql
=> SELECT * FROM pgmq.list_queues();
```
```text
 queue_name |          created_at           | is_partitioned | is_unlogged
------------+-------------------------------+----------------+-------------
 pgmq_queue | 2026-04-10 16:07:46.688943+03 | f              | f
(1 row)
```
Также были созданы таблицы для сообщений очереди: основная q\_pgmq\_queue и архивная a\_pgmq\_queue:
```sql
=> \dt pgmq.*
```
```text
            List of relations
 Schema |     Name     | Type  |  Owner
--------+--------------+-------+---------
 pgmq   | a_pgmq_queue | table | student
 pgmq   | meta         | table | student
 pgmq   | q_pgmq_queue | table | student
(3 rows)
```
Поместим в очередь несколько сообщений (полезная информация представляется значением типа jsonb)...
```sql
=> SELECT pgmq.send('pgmq_queue', to_jsonb(i))
FROM (
    VALUES ('alpha'), ('beta'), ('gamma')
) AS v(i);
```
```text
 send
------
    1
    2
    3
(3 rows)
```
...и заглянем в основную таблицу очереди:
```sql
=> SELECT msg_id, enqueued_at, message
FROM pgmq.q_pgmq_queue
ORDER BY msg_id;
```
```text
 msg_id |          enqueued_at          | message
--------+-------------------------------+---------
      1 | 2026-04-10 16:07:47.207955+03 | "alpha"
      2 | 2026-04-10 16:07:47.207955+03 | "beta"
      3 | 2026-04-10 16:07:47.207955+03 | "gamma"
(3 rows)
```

## Page 8: Page 8 {#source-page-008}

![Source slide page 8: Page 8](images/page-008.png){#source-figure-008 caption="Source slide page 8: Page 8"}

Простой способ забрать сообщение из очереди — вызвать функцию pop:
```sql
=> SELECT msg_id, enqueued_at, message
FROM pgmq.pop('pgmq_queue');
```
```text
 msg_id |          enqueued_at          | message
--------+-------------------------------+---------
      1 | 2026-04-10 16:07:47.207955+03 | "alpha"
(1 row)
```
Другие обработчики тоже могут брать сообщения:
```sh
student$ psql -d ext_async
```
```sql
=> SELECT msg_id, enqueued_at, message
FROM pgmq.pop('pgmq_queue');
```
```text
 msg_id |          enqueued_at          | message
--------+-------------------------------+---------
      2 | 2026-04-10 16:07:47.207955+03 | "beta"
(1 row)
```
```sh
student$ psql -d ext_async
```
```sql
=> SELECT msg_id, enqueued_at, message
FROM pgmq.pop('pgmq_queue');
```
```text
 msg_id |          enqueued_at          | message
--------+-------------------------------+---------
      3 | 2026-04-10 16:07:47.207955+03 | "gamma"
(1 row)
```
А первый при очередном обращении обнаружит, что очередь пуста:
```sql
=> SELECT msg_id, enqueued_at, message
FROM pgmq.pop('pgmq_queue');
```
```text
 msg_id | enqueued_at | message
--------+-------------+---------
(0 rows)
```
И напоследок удалим саму очередь. При этом ее основная и архивная таблицы исчезнут, как и информация в таблице meta:
```sql
=> SELECT pgmq.drop_queue('pgmq_queue');
```
```text
 drop_queue
------------
 t
(1 row)
```
```sql
=> \dt pgmq.*
```
```text
        List of relations
 Schema | Name | Type  |  Owner
--------+------+-------+---------
 pgmq   | meta | table | student
(1 row)
```

## Page 9: Очередь своими руками {#source-page-009}

**Возможные плюсы**
  - не требуются внешние зависимости
  - простые требования — простая реализация
- Минусы
  - требуется отладка и тестирование
  - при усложнении требований готовая система может обойтись дешевле

### Комментарии преподавателя

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

Конечно, нужно понимать, что:

● реализация должна быть сделана аккуратно, иначе она может привести к проблемам эксплуатации;

● если к системе очередей предъявляются серьезные требования (или есть шанс, что такие требования появятся в будущем), то развитие, тестирование и поддержка собственного решения, наоборот, может оказаться невыгодной.

Далее мы посмотрим, как реализовать очередь сообщений в PostgreSQL своими руками, и какие подводные камни есть на этом пути.

## Page 10: Реализация очереди сообщений {#source-page-010}

Наша задача: реализовать простую очередь сообщений с возможностью конкурентного получения сообщений из нескольких процессов. Полезную информацию снова представим типом JSON — так очередь будет достаточно универсальна.
В каждый конкретный момент времени в таблице сообщений не будет много строк, но за все время работы их может оказаться существенное количество. Поэтому идентификатор надо сразу сделать 64-разрядным:
```sql
=> CREATE TABLE msg_queue(
    id bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
    payload jsonb NOT NULL,
    pid integer DEFAULT NULL -- процесс-обработчик
);
```
```sql
CREATE TABLE
```
Вставка сообщений в очередь проста:
```sql
=> INSERT INTO msg_queue(payload)
VALUES
    (to_jsonb(1)),
    (to_jsonb(2)),
    (to_jsonb(3));
```
```sql
INSERT 0 3
```
Теперь займемся функцией получения и блокирования очередного сообщения.
Нам требуется блокировать полученную строку, чтобы одно сообщение не могло быть выбрано два раза (двумя одновременно работающими обработчиками). Это можно сделать с помощью фразы FOR UPDATE:
```text
=> BEGIN;
```
```text
BEGIN
```
```sql
=> SELECT * FROM msg_queue
WHERE pid IS NULL -- никем не обрабатывается
ORDER BY id LIMIT 1 -- одно в порядке поступления
FOR UPDATE;
```
```text
 id | payload | pid
----+---------+-----
  1 | 1       |
(1 row)
```
Но в таком случае аналогичный запрос в другом процессе будет заблокирован до завершения первой транзакции.
```sql
=> \c ext_async
```
```text
You are now connected to database "ext_async" as user "student".
```
```text
=> BEGIN;
```
```text
BEGIN
```
```sql
=> SELECT * FROM msg_queue
WHERE pid IS NULL
ORDER BY id LIMIT 1
FOR UPDATE;
```
Вторая транзакция заблокирована.
```sql
=> DELETE FROM msg_queue
WHERE id = 1;
```
```sql
DELETE 1
```
```text
=> COMMIT;
```
```text
COMMIT
```
```text
 id | payload | pid
----+---------+-----
  2 | 2       |
(1 row)
```

## Page 11: Page 11 {#source-page-011}

```text
=> COMMIT;
```
```text
COMMIT
```
Для того чтобы не останавливаться на заблокированных строках, служит фраза SKIP LOCKED команды SELECT.
```text
=> BEGIN;
```
```text
BEGIN
```
```sql
=> SELECT * FROM msg_queue
WHERE pid IS NULL
ORDER BY id LIMIT 1
FOR UPDATE SKIP LOCKED;
```
```text
 id | payload | pid
----+---------+-----
  2 | 2       |
(1 row)
```
```text
=> BEGIN;
```
```text
BEGIN
```
```sql
=> SELECT * FROM msg_queue
WHERE pid IS NULL
ORDER BY id LIMIT 1
FOR UPDATE SKIP LOCKED;
```
```text
 id | payload | pid
----+---------+-----
  3 | 3       |
(1 row)
```
```text
=> COMMIT;
```
```text
COMMIT
```
```text
=> COMMIT;
```
```text
COMMIT
```
Итак, функция для получения и блокирования очередного сообщения может выглядеть следующим образом:
```sql
=> CREATE FUNCTION take_message(OUT msg msg_queue)
LANGUAGE sql VOLATILE
BEGIN ATOMIC
    UPDATE msg_queue
    SET pid = pg_backend_pid()
    WHERE id = (SELECT id FROM msg_queue
WHERE pid IS NULL
        ORDER BY id LIMIT 1
FOR UPDATE SKIP LOCKED) RETURNING *;
END;
```
```sql
CREATE FUNCTION
```
В практических заданиях к темам «Очистка» и «Фоновые задания» мы рассматривали типичное решение для получения пакета строк таблицы, например, с целью обновления или удаления. Запрос выглядел так:
```sql
WITH batch AS (
    SELECT * FROM t
    WHERE /* необходимые условия */
    LIMIT /* размер пакета */
    FOR UPDATE SKIP LOCKED
)
...
```
Как видите, в обоих случаях используется тот же самый подход: выбирается и блокируется часть строк (одна или несколько), при этом уже заблокированные строки пропускаются.
Теперь напишем функцию завершения работы с сообщением. Мы будем просто удалять его из очереди.
```sql
=> CREATE FUNCTION complete_message(msg msg_queue) RETURNS void
LANGUAGE sql VOLATILE
BEGIN ATOMIC
  DELETE FROM msg_queue WHERE id = msg.id;
END;
```
```sql
CREATE FUNCTION
```

## Page 12: Page 12 {#source-page-012}

Теперь мы готовы написать цикл обработки сообщений. Оформим его в виде процедуры.
```sql
=> CREATE PROCEDURE process_queue() AS $$
DECLARE
    msg msg_queue;
BEGIN
    LOOP
        SELECT * INTO msg FROM take_message();
        EXIT WHEN msg.id IS NULL;
```
```text
        -- обработка
        PERFORM pg_sleep(1);
        RAISE NOTICE '[%] processed %; n_tup_del=%, backend_xmin=%',
            pg_backend_pid(),
            msg.payload,
```
```sql
            (SELECT n_tup_del FROM pg_stat_xact_all_tables  -- статистика, накопленная внутри транзакции
             WHERE relname = 'msg_queue'),
```
```sql
            (SELECT backend_xmin FROM pg_stat_activity
             WHERE pid = pg_backend_pid());
```
```text
        PERFORM complete_message(msg);
    END LOOP;
END;
$$ LANGUAGE plpgsql;
```
```sql
CREATE PROCEDURE
```
В этом варианте цикл заканчивается, когда в очереди не остается необработанных сообщений. Вместо этого можно не прекращать цикл, но продолжать ожидать новые события, засыпая, например, на одну секунду.
Пробуем.
```text
=> CALL process_queue();
```
```text
NOTICE:  [35350] processed 2; n_tup_del=0, backend_xmin=833
NOTICE:  [35350] processed 3; n_tup_del=1, backend_xmin=833
CALL
```
Теперь в два потока.
```sql
=> INSERT INTO msg_queue(payload)
SELECT to_jsonb(id) FROM generate_series(1,10) id;
```
```sql
INSERT 0 10
```
```text
=> \timing on
```
```text
Timing is on.
```
```text
=> CALL process_queue();
```
```text
=> CALL process_queue();
```
```text
NOTICE:  [36175] processed 2; n_tup_del=0, backend_xmin=835
NOTICE:  [36175] processed 4; n_tup_del=1, backend_xmin=835
NOTICE:  [36175] processed 6; n_tup_del=2, backend_xmin=835
NOTICE:  [36175] processed 8; n_tup_del=3, backend_xmin=835
NOTICE:  [36175] processed 10; n_tup_del=4, backend_xmin=835
CALL
```
```text
NOTICE:  [35350] processed 1; n_tup_del=0, backend_xmin=835
NOTICE:  [35350] processed 3; n_tup_del=1, backend_xmin=835
NOTICE:  [35350] processed 5; n_tup_del=2, backend_xmin=835
NOTICE:  [35350] processed 7; n_tup_del=3, backend_xmin=835
NOTICE:  [35350] processed 9; n_tup_del=4, backend_xmin=835
CALL
Time: 5019,651 ms (00:05,020)
```
```text
=> \timing off
```
```text
Timing is off.
```
Обработка 10 сообщений двумя потоками заняла около 5 секунд, но горизонт транзакций держался на одном уровне все время обработки очереди! Это будет мешать выполнению очистки и создавать проблемы для всей базы данных.

## Page 13: Вспоминаем про горизонт {#source-page-013}

**Что получилось: одна большая транзакция**
  - take\_message();
  - - обработка
  - complete\_message();
  - take\_message();
  - - обработка
  - complete\_message();
  - take\_message();
  - - обработка
  - complete\_message();
  - COMMIT;

### Комментарии преподавателя

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

## Page 14: Вспоминаем про горизонт {#source-page-014}

**Что надо: каждое событие в отдельной транзакции**
  - take\_message();
  - - обработка
  - complete\_message();
  - take\_message();
  - - обработка
  - complete\_message();
  - take\_message();
  - - обработка
  - complete\_message();
  - COMMIT;
  - COMMIT;
  - COMMIT;

### Комментарии преподавателя

Чтобы таких проблем не возникало, надо раздробить длинную транзакцию на несколько более коротких. В нашем случае — обрабатывать каждое событие в собственной транзакции.

## Page 15: Вспоминаем про горизонт {#source-page-015}

![Source slide page 15: Вспоминаем про горизонт](images/page-015.png){#source-figure-015 caption="Source slide page 15: Вспоминаем про горизонт"}

**Еще лучше: позволить обработке события состоять из нескольких транзакций**

### Комментарии преподавателя

Более того, обработка одного события тоже может (в принципе) разбиваться на несколько транзакций.

В таком случае мы сначала фиксируем изменение статуса события в очереди («в работе»), затем выполняем обработку, а в конце фиксируем факт завершения работы с событием (например, удаляем его из таблицы).

## Page 16: Учитываем горизонт транзакций {#source-page-016}

Это легко сделать, поскольку процедура позволяет управлять транзакциями.
```sql
=> CREATE OR REPLACE PROCEDURE process_queue() AS $$
DECLARE
    msg msg_queue;
BEGIN
    LOOP
        SELECT * INTO msg FROM take_message();
        COMMIT; --<<
        EXIT WHEN msg.id IS NULL;
```
```text
        -- обработка
        PERFORM pg_sleep(1);
        RAISE NOTICE '[%] processed %; n_dead_tup=%, n_tup_del=%, backend_xmin=%',
            pg_backend_pid(),
            msg.payload,
```
```sql
            (SELECT n_dead_tup FROM pg_stat_all_tables  -- статистика, учитываемая автоочисткой
             WHERE relname = 'msg_queue'),
```
```sql
            (SELECT n_tup_del FROM pg_stat_xact_all_tables
             WHERE relname = 'msg_queue'),
```
```sql
            (SELECT backend_xmin FROM pg_stat_activity
             WHERE pid = pg_backend_pid());
```
```text
        PERFORM complete_message(msg);
        COMMIT; --<<
    END LOOP;
END;
$$ LANGUAGE plpgsql;
```
```sql
CREATE PROCEDURE
```
Проверим:
```sql
=> INSERT INTO msg_queue(payload)
SELECT to_jsonb(id) FROM generate_series(1,5) id;
```
```sql
INSERT 0 5
```
```text
=> CALL process_queue();
```
```text
NOTICE:  [35350] processed 1; n_dead_tup=0, n_tup_del=0, backend_xmin=840
NOTICE:  [35350] processed 2; n_dead_tup=0, n_tup_del=1, backend_xmin=842
NOTICE:  [35350] processed 3; n_dead_tup=0, n_tup_del=2, backend_xmin=844
NOTICE:  [35350] processed 4; n_dead_tup=0, n_tup_del=3, backend_xmin=846
NOTICE:  [35350] processed 5; n_dead_tup=0, n_tup_del=4, backend_xmin=848
CALL
```
Теперь горизонт транзакций продвигается вперед и не мешает очистке. Однако момент срабатывания автоматической очистки определяется на основе данных статистики, которая обновляется лишь по окончании работы всего оператора CALL.
```sql
=> SELECT n_dead_tup, n_live_tup, n_mod_since_analyze, n_ins_since_vacuum
FROM pg_stat_all_tables
WHERE relname = 'msg_queue';
```
```text
 n_dead_tup | n_live_tup | n_mod_since_analyze | n_ins_since_vacuum
------------+------------+---------------------+--------------------
         10 |          0 |                  15 |                  5
(1 row)
```
Таким образом, чтобы таблица с очередью вовремя очищалась, можно:
модифицировать процедуру process\_queue таким образом, чтобы обеспечить ее гарантированное периодическое завершение и положиться на автоочистку; периодически запускать обычную (неавтоматическую) очистку. Один из способов реализации — использование фоновых процессов, которые рассмотрены в соответствующей теме этого курса.

## Page 17: Чего не хватает {#source-page-017}

![Source slide page 17: Чего не хватает](images/page-017.png){#source-figure-017 caption="Source slide page 17: Чего не хватает"}

**Зависшие сообщения**
  - остаются в статусе «в работе» при аварийном завершении обработчика
- Решение
  - проверять существование процесса-обработчика, указанного в таблице
  - при отсутствии возвращать сообщение в статус «новый» (возможно)

### Комментарии преподавателя

Чего не хватает в нашей реализации?

Во-первых, обработчик может завершиться аварийно. Если мы фиксируем изменение статуса обработки, то событие «повиснет» в статусе «в работе» и не будет больше обрабатываться.

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

Что делать в таком случае? Если обработка события выполнялась в одной транзакции, то она была прервана и, следовательно, можно безопасно вернуть событие в статус «новое» — оно будет обработано повторно.

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

## Page 18: Чего не хватает {#source-page-018}

**Корректная обработка исключительных ситуаций**
- Сохранение результатов обработки
- Решение
  - не удалять обработанные сообщения, а помечать отдельным статусом
  - потребуется правильный индекс
  - потребуется периодическое удаление исторических данных
  - обращаем внимание на очистку

### Комментарии преподавателя

Во-вторых, наша реализация никак не обрабатывает исключительные ситуации. Это, конечно, несложно добавить. При возникновении исключения хотелось бы иметь информацию о том, что случилось.

Да и если событие обработано без ошибок, может быть полезным сохранять какую-то информацию об обработке. Это, конечно, зависит от конкретной задачи.

Наша реализация удаляет обработанные события из очереди, но вместо этого можно оставлять их, помечая специальным статусом («завершено», «ошибка» и т. п.). Тогда всю информацию об обработке можно иметь непосредственно в таблице с событиями. В таком случае потребуется эффективный доступ к еще не обработанным сообщениям: частичный индекс с условием pid IS NULL. (Другим решением может быть перенос обработанных событий в отдельную таблицу.)

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

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

## Page 19: Итоги {#source-page-019}

**Асинхронная обработка полезна во многих случаях**
- Внешние системы имеет смысл использовать, если
  - вписываются в общую архитектуру информационной системы
  - к реализации предъявляются серьезные требования
- Очередь сообщений в базе данных — простое решение для простых задач
  - важна правильная реализация:
  - эффективное получение очередного события (SKIP LOCKED),
  - избегание долгих транзакций,
  - своевременная очистка таблицы очереди
  - чем больше требований, тем сложнее будет реализация

## Page 20: Практика {#source-page-020}

1. В приложении предусмотрен механизм фоновых заданий, но серверная часть обработки очереди отсутствует.
- Напишите недостающие функции: - take\_task — получает очередное задание из очереди; - complete\_task — завершает обработку задания; - process\_tasks — основной цикл обработки заданий. 2. Запустите процедуру обработки очереди заданий в фоновом режиме. Проверьте, что фоновые задания, поставленные в очередь в приложении, выполняются, а результаты их работы доступны для просмотра.

### Комментарии преподавателя

1. Фоновые задания позволяют запустить специально зарегистрированную функцию из пользовательского интерфейса и затем просматривать состояние и результат выполнения.

В качестве результата функция может возвращать множество строк, то есть в простейшем виде тело функции может состоять из одного SQL-запроса. На вход функция должна принимать один параметр типа jsonb. Пример задания: public.greeting\_program.

Напишите подпрограммы take\_task, complete\_task и process\_tasks по аналогии с показанными в демонстрации примерами. Учтите:

● take\_task должна возвращать задачу в статусе «scheduled» и заполнить подходящие поля таблицы tasks:

started = текущее время, status = «running», pid = номер процесса.

● complete\_task должна не удалять задание, а заполнить поля tasks:

finished = текущее время, при нормальном завершении: status = «finished», result = результат, в случае ошибки: status = «error», result = сообщение об ошибке.

● process\_tasks не должна завершаться; организуйте бесконечный цикл с задержкой в 1 сек между задачами. Убедитесь, что в режиме ожидания не возникает долгой транзакции. Для удобства установите параметр application\_name в значение «process\_tasks».

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

## Page 21: 1. Реализация обработки очереди заданий {#source-page-021}

Функция получения задания из очереди аналогична показанной в демонстрации, но должна учитывать поля таблицы:
```sql
=> SELECT * FROM tasks \gx
```
```text
-[ RECORD 1 ]---------
task_id    | 1
program_id | 1
status     | scheduled
params     |
pid        |
started    |
finished   |
result     |
host       |
port       |
```
(Игнорируйте столбцы host и port — они пригодятся в теме «Обзор физической репликации».)
```sql
=> CREATE FUNCTION take_task(OUT task tasks) AS $$
BEGIN
    SELECT *
    INTO task
    FROM tasks
    WHERE status = 'scheduled'
    ORDER BY task_id LIMIT 1
    FOR UPDATE SKIP LOCKED;
```
```sql
    UPDATE tasks
    SET status = 'running',
        started = current_timestamp,
        pid = pg_backend_pid()
    WHERE task_id = task.task_id;
END;
$$ LANGUAGE plpgsql VOLATILE;
```
```sql
CREATE FUNCTION
```
Поскольку мы не будем удалять задания из очереди, создадим частичный индекс для эффективного доступа к следующему необработанному заданию:
```sql
=> CREATE INDEX ON tasks(task_id) WHERE status = 'scheduled';
```
```sql
CREATE INDEX
```
Функция завершения работы с заданием дополнительно принимает статус завершения и текстовый результат:
```sql
=> CREATE FUNCTION complete_task(task tasks, status text, result text) RETURNS void
LANGUAGE sql VOLATILE
BEGIN ATOMIC
    UPDATE tasks
    SET finished = current_timestamp,
        status = complete_task.status,
        result = complete_task.result
    WHERE task_id = task.task_id;
END;
```
```sql
CREATE FUNCTION
```
Процедура обработки очереди:

## Page 22: Page 22 {#source-page-022}

```sql
=> CREATE PROCEDURE process_tasks() AS $$
DECLARE
    task tasks;
    result text;
    ctx text;
BEGIN
    SET application_name = 'process_tasks';
    <<forever>>
    LOOP
        PERFORM pg_sleep(1);
        SELECT * INTO task FROM take_task();
        COMMIT;
        CONTINUE forever WHEN task.task_id IS NULL;
```
```text
        BEGIN
```
```text
            result := empapi.run(task);
            PERFORM complete_task(task, 'finished', result);
        EXCEPTION
            WHEN others THEN
                GET STACKED DIAGNOSTICS
                    result := MESSAGE_TEXT, ctx := PG_EXCEPTION_CONTEXT;
                PERFORM complete_task(
                    task, 'error', result || E'\n' ||  ctx
                );
        END;
```
```text
        COMMIT;
    END LOOP;
END;
$$ LANGUAGE plpgsql;
```
```sql
CREATE PROCEDURE
```
Обратите внимание, что первая команда COMMIT предшествует команде CONTINUE. В противном случае при отсутствии заданий возникала бы долгая транзакция.
Несколько слов о том, зачем нужна функция run. В принципе, выполнить задание и получить результат можно было бы таким образом:
```sql
func := (
    SELECT p.func FROM programs p WHERE p.program_id = task.program_id
);
```
```sql
EXECUTE format(
    $$SELECT string_agg(f::text, E'\n') FROM %I($1) AS f$$,
    func
)
INTO result
USING task.params;
```
К сожалению, PL/pgSQL не позволяет гибко работать со значениями составного типа: у значения неизвестного наперед типа (record) нельзя перебрать все имеющиеся в нем поля. Поэтому для вывода приходится полагаться на стандартное преобразование строки в текст. Это будет некрасиво выглядеть в случае нескольких полей:
```sql
=> SELECT string_agg(f::text, E'\n') FROM greeting_task() AS f;
```
```text
     string_agg
---------------------
 (1,"Hello, world!")+
 (2,"Hello, world!")+
 (3,"Hello, world!")
(1 row)
```
Для аккуратного оформления результата можно воспользоваться другим процедурным языком. Мы используем функцию, написанную на PL/Python. Функция run не вызывается напрямую приложением, но в теме «Обзор физической репликации» мы будем вызывать ее на другом сервере, поэтому она находится в схеме empapi, а не public.
Подробнее о том, в каких случаях могут пригодиться другие языки, будет рассказано в теме «Языки программирования».
### 2. Запуск обработки очереди в фоновом режиме
В очереди стоит тестовое задание:
```sql
=> SELECT * FROM tasks \gx
```

## Page 23: Page 23 {#source-page-023}

```text
-[ RECORD 1 ]---------
task_id    | 1
program_id | 1
status     | scheduled
params     |
pid        |
started    |
finished   |
result     |
host       |
port       |
```
Запускаем обработку (в один поток) и, если все сделано правильно, оно будет выполнено.
```sql
=> SELECT pg_background_launch('CALL process_tasks()');
```
```text
 pg_background_launch
----------------------
                78423
(1 row)
```
```sql
=> SELECT pg_background_detach(78423);
```
```text
 pg_background_detach
----------------------
(1 row)
```
Между вызовами нужна небольшая задержка, чтобы функция pg\_background\_detach не освободила общую память до того, как фоновый процесс обратится к ней — такова особенность текущей реализации. Поэтому мы не вызываем обе функции в одном операторе.
Подождем немного...
```sql
=> SELECT * FROM tasks \gx
```
```text
-[ RECORD 1 ]-----------------------------
task_id    | 1
program_id | 1
status     | finished
params     |
pid        | 78423
started    | 2026-04-10 16:28:16.530634+03
finished   | 2026-04-10 16:28:17.540188+03
result     | num    greeting              +
           | --- -------------            +
           | 1   Hello, world!            +
           | 2   Hello, world!            +
           | 3   Hello, world!            +
           |
host       |
port       |
```
Задание успешно выполнено. Обратите внимание, что результат выполнения содержит и названия столбцов из оригинального запроса.
Фоновые процессы, обрабатывающие очередь, легко найти благодаря тому, что процедура устанавливает параметр application\_name:
```sql
=> SELECT pid, wait_event_type, wait_event, query
FROM pg_stat_activity
WHERE application_name = 'process_tasks' \gx
```
```text
-[ RECORD 1 ]---+---------------------
pid             | 78423
wait_event_type | Timeout
wait_event      | PgSleep
query           | CALL process_tasks()
```

## Page 24: Практика+ {#source-page-024}

1. Напишите тест, проверяющий, что обработка очереди, показанная в демонстрации, работает корректно при выполнении в несколько потоков. Убедитесь, что тест не проходит, если убрать предложение FOR UPDATE SKIP LOCKED. 2. Добавьте в реализацию проверку «зависших» сообщений. Если такая ситуация будет обнаружена, зависшее сообщение должно быть снова принято в работу.

### Комментарии преподавателя

1. Вставьте в таблицу сообщений большое количество строк и проверьте, что:

а) было обработано каждое сообщение;

б) каждое сообщение было обработано ровно один раз.

Уберите из реализации секундную задержку (имитацию работы), чтобы тест выполнялся быстрее и с достаточным уровнем конкурентности между процессами.

## Page 25: 1. Тестирование реализации очереди {#source-page-025}

```sql
=> CREATE DATABASE ext_async;
```
```sql
CREATE DATABASE
```
```sql
=> \c ext_async
```
```text
You are now connected to database "ext_async" as user "student".
```
Повторим реализацию очереди, показанную в демонстрации.
Таблица:
```sql
=> CREATE TABLE msg_queue(
  id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  payload jsonb NOT NULL,
  pid integer DEFAULT NULL
);
```
```sql
CREATE TABLE
```
Функция получения и блокирования очередного сообщения:
```sql
=> CREATE FUNCTION take_message(OUT msg msg_queue) AS $$
BEGIN
    SELECT *
    INTO msg
    FROM msg_queue
    WHERE pid IS NULL
    ORDER BY id LIMIT 1
    FOR UPDATE SKIP LOCKED;
```
```sql
    UPDATE msg_queue
    SET pid = pg_backend_pid()
    WHERE id = msg.id;
END;
$$ LANGUAGE plpgsql VOLATILE;
```
```sql
CREATE FUNCTION
```
Функция завершения работы с сообщением:
```sql
=> CREATE FUNCTION complete_message(msg msg_queue) RETURNS void
LANGUAGE sql VOLATILE
BEGIN ATOMIC
    DELETE FROM msg_queue WHERE id = msg.id;
END;
```
```sql
CREATE FUNCTION
```
В процедуру обработки очереди внесем изменение: вместо секундной задержки будем записывать информацию об обрабатываемом сообщении в отдельную таблицу:
```sql
=> CREATE TABLE msg_log(
    id bigint,
    pid integer
);
```
```sql
CREATE TABLE
```
```sql
=> CREATE PROCEDURE process_queue() AS $$
DECLARE
    msg msg_queue;
BEGIN
    LOOP
        SELECT * INTO msg FROM take_message();
        EXIT WHEN msg.id IS NULL;
        COMMIT;
```
```sql
        -- обработка
        INSERT INTO msg_log(id, pid) VALUES (msg.id, pg_backend_pid());
```
```text
        PERFORM complete_message(msg);
        COMMIT;
    END LOOP;
END;
$$ LANGUAGE plpgsql;
```

## Page 26: Page 26 {#source-page-026}

```sql
CREATE PROCEDURE
```
Создаем большое количество сообщений:
```sql
=> INSERT INTO msg_queue(payload)
SELECT to_jsonb(id) FROM generate_series(1,1000) id;
```
```sql
INSERT 0 1000
```
Запускаем обработку в два потока, засекая время:
```sh
student$ psql ext_async
```
```text
=> \timing on
```
```text
Timing is on.
```
```text
=> CALL process_queue();
```
```text
=> CALL process_queue();
```
```text
CALL
```
```text
CALL
Time: 3545,203 ms (00:03,545)
```
```text
=> \timing off
```
```text
Timing is off.
```
Проанализируем результаты. При корректной работе мы должны обнаружить в журнальной таблице ровно 1000 уникальных идентификаторов, что будет означать, что обработаны все события, и ни одно не обработано дважды.
```sql
=> SELECT count(*), count(DISTINCT id) FROM msg_log;
```
```text
 count | count
-------+-------
  1000 |  1000
(1 row)
```
Все корректно.
Проверим теперь реализацию без предложения FOR UPDATE SKIP LOCKED.
```sql
=> CREATE OR REPLACE FUNCTION take_message(OUT msg msg_queue) AS $$
BEGIN
    SELECT *
    INTO msg
    FROM msg_queue
    WHERE pid IS NULL
    ORDER BY id LIMIT 1
    /*FOR UPDATE SKIP LOCKED*/;
```
```sql
    UPDATE msg_queue
    SET pid = pg_backend_pid()
    WHERE id = msg.id;
END;
$$ LANGUAGE plpgsql VOLATILE;
```
```sql
CREATE FUNCTION
```
```text
=> TRUNCATE msg_queue;
```
```text
TRUNCATE TABLE
```
```text
=> TRUNCATE msg_log;
```
```text
TRUNCATE TABLE
```
```sql
=> INSERT INTO msg_queue(payload)
SELECT to_jsonb(id) FROM generate_series(1,1000) id;
```
```sql
INSERT 0 1000
```
Запускаем обработку:
```text
=> \timing on
```
```text
Timing is on.
```
```text
=> CALL process_queue();
```
```text
=> CALL process_queue();
```

## Page 27: Page 27 {#source-page-027}

```text
CALL
Time: 6964,778 ms (00:06,965)
```
```text
CALL
```
```text
=> \timing off
```
```text
Timing is off.
```
```sql
=> SELECT count(*), count(DISTINCT id) FROM msg_log;
```
```text
 count | count
-------+-------
  1713 |  1000
(1 row)
```
Как видим, часть сообщений была обработана дважды. Например:
```sql
=> SELECT id, array_agg(pid) FROM msg_log
GROUP BY id HAVING count(*) > 1
LIMIT 10;
```
```text
  id  |   array_agg
------+---------------
 1009 | {79195,78975}
 1010 | {78975,79195}
 1012 | {78975,79195}
 1013 | {79195,78975}
 1015 | {78975,79195}
 1017 | {79195,78975}
 1019 | {79195,78975}
 1021 | {79195,78975}
 1022 | {79195,78975}
 1024 | {78975,79195}
(10 rows)
```
Это произошло из-за того, что сообщение, обрабатываемое одним процессом, никак не блокируется и доступно для другого процесса.
Восстановим корректную функцию:
```sql
=> CREATE OR REPLACE FUNCTION take_message(OUT msg msg_queue) AS $$
BEGIN
    SELECT *
    INTO msg
    FROM msg_queue
    WHERE pid IS NULL
    ORDER BY id LIMIT 1
    FOR UPDATE SKIP LOCKED;
```
```sql
    UPDATE msg_queue
    SET pid = pg_backend_pid()
    WHERE id = msg.id;
END;
$$ LANGUAGE plpgsql VOLATILE;
```
```sql
CREATE FUNCTION
```
### 2. Обработка зависших сообщений
Мы можем перехватить ошибку, возникающую при обработке события, но тем не менее всегда есть шанс того, что сама процедура-обработчик завершится аварийно. Сымитируем такую ситуацию:
```text
=> TRUNCATE msg_queue;
```
```text
TRUNCATE TABLE
```
```text
=> TRUNCATE msg_log;
```
```text
TRUNCATE TABLE
```
```sql
=> INSERT INTO msg_queue(payload)
SELECT to_jsonb(id) FROM generate_series(1,1000) id;
```
```sql
INSERT 0 1000
```
Запускаем обработку...
```text
=> CALL process_queue();
```

## Page 28: Page 28 {#source-page-028}

...а в это время в другом сеансе:
```text
=> BEGIN;
```
```text
BEGIN
```
```text
=> LOCK TABLE msg_log;
```
```text
LOCK TABLE
```
```sql
=> SELECT pid, pg_terminate_backend(pid) FROM msg_log LIMIT 1;
```
```text
  pid  | pg_terminate_backend
-------+----------------------
 79195 | t
(1 row)
```
```text
=> COMMIT;
```
```text
COMMIT
```
```sql
FATAL:  terminating connection due to administrator command
CONTEXT:  SQL statement "INSERT INTO msg_log(id, pid) VALUES (msg.id, pg_backend_pid())"
PL/pgSQL function process_queue() line 11 at SQL statement
server closed the connection unexpectedly
This probably means the server terminated abnormally
before or while processing the request.
connection to server was lost
```
Обработчик «упал». Причем, благодаря команде LOCK TABLE, — сразу после того, как зафиксировал номер процесса в таблице очереди. В очереди остались необработанные сообщения и среди них — одно зависшее:
```sql
=> SELECT count(*), count(DISTINCT id) FROM msg_log;
```
```text
 count | count
-------+-------
    32 |    32
(1 row)
```
```sql
=> SELECT * FROM msg_queue WHERE pid IS NOT NULL;
```
```text
  id  | payload |  pid
------+---------+-------
 2033 | 33      | 79195
(1 row)
```
Самый простой способ исправить ситуацию — изменить функцию выбора сообщения:
```sql
=> CREATE OR REPLACE FUNCTION take_message(OUT msg msg_queue) AS $$
BEGIN
    SELECT *
    INTO msg
    FROM msg_queue
    WHERE pid IS NULL OR pid NOT IN (SELECT pid FROM pg_stat_activity)
    ORDER BY id LIMIT 1
    FOR UPDATE SKIP LOCKED;
```
```sql
    UPDATE msg_queue
    SET pid = pg_backend_pid()
    WHERE id = msg.id;
END;
$$ LANGUAGE plpgsql VOLATILE;
```
```sql
CREATE FUNCTION
```
Если события обрабатываются быстро и важна высокая пропускная способность, то проверку лучше выполнять отдельно и только время от времени, чтобы избежать постоянного обращения к pg\_stat\_activity.
Снова запустим обработчик, и все сообщения, включая зависшее, будут обработаны.
```text
=> CALL process_queue();
```
```text
CALL
```
```sql
=> SELECT count(*), count(DISTINCT id) FROM msg_log;
```

## Page 29: Page 29 {#source-page-029}

```text
 count | count
-------+-------
  1000 |  1000
(1 row)
```