Для организации обработки потока задач используются очереди. Они нужны для накопления и распределения задач по исполнителям. Также очереди могут обеспечивать дополнительные требования к обработке задач: гарантия доставки, гарантия однократного исполнения, приоритезация и т. д.

Как правило, используются готовые системы очередей сообщений (MQ — message queue), но иногда нужно организовать ad hoc очередь или какую-нибудь специализированную (например, очередь с приоритетом и отложенным перезапуском не обработанных из-за исключений задач). О создании таких очередей и пойдёт речь ниже.

Ограничения применимости

Предлагаемые решения предназначены для обработки потока однотипных задач. Они не подходят для организации pub/sub или обмена сообщениями между слабо связанными системами и компонентами.

Очередь поверх реляционной БД хорошо работает при малых и средних нагрузках (сотни тысяч задач в сутки, десятки-сотни исполнителей), но для больших потоков лучше использовать специализированное решение.

Суть метода в пяти словах

select ... for update skip locked

Базовая очередь

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

Таблица для простейшей очереди содержит саму задачу и её статус:

create table task
(
    id          bigint not null primary key,
    status      integer not null default 0      -- 0 - новая, 1 - в работе, 2 - выполнена
);

create index task__status__idx on task (status);

Добавление задачи:

insert into task (id) values ($1) on conflict (id) do nothing;

Получение следующей задачи:

with next_task as (
    select id from task
    where status = 0
    limit 1
    for update skip locked
)
update task
set
    status = 1
from next_task
where task.id = next_task.id
returning task.id;

Завершение задачи:

update task
set
    status = 2
where id = $1;

Очередь с приоритетами

В простом случае id задачи является её приоритетом. Меняется только запрос на получение следующей задачи — добавляется условие сортировки order by id с требуемым порядком обработки задач. Также нужно создать составной индекс по (status, id).

Либо для приоритета добавляется отдельный столбец:

create table task
(
    id          bigint not null primary key,
    priority    integer not null,
    status      integer not null default 0      -- 0 - новая, 1 - в работе, 2 - выполнена
);

create index task__status__priority__idx on task (status, priority);

insert into task (id, priority) values ($1, $2) on conflict (id) do nothing;

with next_task as (
    select id from task
    where status = 0
    order by priority
    limit 1
    for update skip locked
)
update task
set
    status = 1
from next_task
where task.id = next_task.id
returning task.id;

Выделенный столбец позволяет менять приоритет задачи "на лету".

Очередь с повтором "упавших" задач

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

create table task
(
    id          bigint not null primary key,
    status      integer not null default 0,     -- 0 - новая, 1 - в работе, 2 - выполнена, 3 - ошибка, 4 - фатальная ошибка (повтора не будет)
    attempt     integer not null default 0,
    delayed_to  timestamp null,
    error_text  text null
);

create index task__status__delayed_to__idx on task (status, delayed_to);

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

• attempt — номер попытки; нужен для принятия решения о необходимости повтора (ограничение количества попыток) и для выбора задержки перед повтором (например, каждая следующая попытка откладывается на 10 * attempt минут);
• delayed_to — время следующей попытки выполнения задачи;
• error_text — текст ошибки.

Текст ошибки нужен для группировки по типам ошибки.

Пример. Система мониторинга сообщает, что в очереди скопились тысячи задач со статусом "ошибка". Выполняем запрос:

select error_text, count(*) from task where status = 3 group by 1 order by 2 desc;

За подробностями идём в логи исполнителей. Исправляем ситуацию, вызвавшую ошибки (если это возможно). При необходимости ускоряем перезапуск задач установкой статуса в 0 или сдвигом времени следующей попытки.

with next_task as (
    select id from task
    where status = 0
    limit 1
    for update skip locked
)
update task
set
    status = 1,
    attempt = attempt + 1,
    delayed_to = null,
    error_text = null
from next_task
where task.id = next_task.id
returning task.id;

with next_task as (
    select id from task
    where status = 3
      and delayed_to < localtimestamp
    limit 1
    for update skip locked
)
update task
set
    status = 1,
    attempt = attempt + 1,
    delayed_to = null,
    error_text = null
from next_task
where task.id = next_task.id
returning task.id;

update task
set
    status = 2,
    delayed_to = null,
    error_text = null
where id = $1;

update task
set
    status = 3,
    delayed_to = localtimestamp + make_interval(mins => 5 * attempt),
    error_text = $2
where id = $1;

update task
set
    status = 4,
    delayed_to = null,
    error_text = $2
where id = $1;

Запрос получения следующей задачи разделён на два, чтобы СУБД могла построить эффективный план запроса для очереди с приоритетом. Условие отбора с or может очень плохо сочетаться с сортировкой order by.

Сбор метрик

Добавляем такие атрибуты:

• время создания задачи;
• время изменения задачи;
• время начала и завершения выполнения задачи.

create table task
(
    id          bigint not null primary key,
    status      integer not null default 0,     -- 0 - новая, 1 - в работе, 2 - выполнена, 3 - ошибка, 4 - фатальная ошибка (повтора не будет)
    attempt     integer not null default 0,
    begin_time  timestamp null,
    end_time    timestamp null,
    delayed_to  timestamp null,
    error_text  text null,
    created     timestamp not null default localtimestamp,
    updated     timestamp not null default localtimestamp
);

create index task__status__delayed_to__idx on task (status, delayed_to);
create index task__updated__idx on task (updated);

Учитываем добавленные столбцы во всех запросах.

with next_task as (
    select id from task
    where status = 0
    limit 1
    for update skip locked
)
update task
set
    status = 1,
    attempt = attempt + 1,
    begin_time = localtimestamp,
    end_time = null,
    delayed_to = null,
    error_text = null,
    updated = localtimestamp
from next_task
where task.id = next_task.id
returning task.id;

with next_task as (
    select id from task
    where status = 3
      and delayed_to < localtimestamp
    limit 1
    for update skip locked
)
update task
set
    status = 1,
    attempt = attempt + 1,
    begin_time = localtimestamp,
    end_time = null,
    delayed_to = null,
    error_text = null,
    updated = localtimestamp
from next_task
where task.id = next_task.id
returning task.id;

update task
set
    status = 2,
    end_time = localtimestamp,
    delayed_to = null,
    error_text = null,
    updated = localtimestamp
where id = $1;

update task
set
    status = 3,
    end_time = localtimestamp,
    delayed_to = localtimestamp + make_interval(mins => 5 * attempt),
    error_text = $2,
    updated = localtimestamp
where id = $1;

update task
set
    status = 4,
    end_time = localtimestamp,
    delayed_to = null,
    error_text = $2,
    updated = localtimestamp
where id = $1;

Примеры, для чего это может быть нужно

Поиск и перезапуск повисших задач:

update task
set
    status = 3,
    end_time = localtimestamp,
    delayed_to = localtimestamp,
    error_text = 'hanged',
    updated = localtimestamp
where status = 1
  and updated < localtimestamp - interval '1 hour';

Удаление старых задач:

delete from task
where updated < localtimestamp - interval '30 days';

Статистика по выполнению задач:

select
    date_trunc('hour', end_time),
    count(*),
    sum(end_time - begin_time),
    avg(end_time - begin_time)
from task
where status = 2
  and end_time >= '2019-12-16'
group by 1
order by 1;

Повторный запуск ранее выполненных задач

Например, обновился документ, нужно его переиндексировать для полнотекстового поиска.

create table task
(
    id              bigint not null primary key,
    task_updated_at timestamp not null default localtimstamp,
    status          integer not null default 0,     -- 0 - новая, 1 - в работе, 2 - выполнена, 3 - ошибка, 4 - фатальная ошибка (повтора не будет)
    begin_time      timestamp null,
    end_time        timestamp null,
    delayed_to      timestamp null,
    error_text      text null,
    created         timestamp not null default localtimestamp,
    updated         timestamp not null default localtimestamp
);

Здесь для времени обновления задачи добавлен столбец task_updated_at, но можно было бы использовать поле created.

Добавление или обновление (перезапуск) задачи:

insert into task (id, task_updated_at) values ($1, $2)
on conflict (id) do update
set
    task_updated_at = excluded.task_updated_at,
    status = case when status = 1 then 1 else 0 end,
    delayed_to = null,
    error_text = null,
    updated = localtimestamp
where task_updated_at < excluded.task_updated_at;

Что здесь происходит. Задача становится "новой", если она сейчас не исполняется.

В запросе завершения задачи также будет проверка, была ли она изменена во время исполнения.

Запросы на получение следующей задачи такие же, как в очереди со сбором метрик.

Успешное завершение задачи:

update task
set
    status = case when begin_time >= updated then 2 else 0 end,
    end_time = localtimestamp,
    delayed_to = null,
    error_text = null,
    updated = localtimestamp
where id = $1;

Завершение задачи с ошибкой: в зависимости от задачи. Можно сделать безусловное откладывание перезапуска, можно при обновлении ставить статус "новая".

Pipeline

Задача проходит несколько стадий. Можно для каждой стадии сделать отдельную очередь. А можно в таблицу добавить соответствующий столбец.

Пример на основе базовой очереди, чтобы не загромождать код. Все ранее описанные модификации без проблем применимы и к этой очереди.

create table task
(
    id      bigint not null primary key,
    stage   integer not null default 0,
    status  integer not null default 0
);

create index task__stage__status__idx on task (stage, status);

Получение следующей задачи на заданной стадии:

with next_task as (
    select id from task
    where stage = $1
      and status = 0
    limit 1
    for update skip locked
)
update task
set
    status = 1
from next_task
where task.id = next_task.id
returning task.id;

Завершение задачи с переходом на указанную стадию:

update task
set
    stage = $2,
    status = 2
where id = $1;

Или переход на следующую по порядку стадию:

update task
set
    stage = stage + 1,
    status = 2
where id = $1;

Задачи по расписанию

Это вариация очереди с повтором.

У каждой задачи может быть своё расписание (в простейшем варианте — периодичность запуска).

create table task
(
    id              bigint not null primary key,
    period          integer not null,               -- периодичность запуска в секундах
    status          integer not null default 0,     -- 0 - новая, 1 - в работе
    next_run_time   timestamp not null default localtimestamp
);

create index task__status__next_run_time__idx on task (status, next_run_time);

Добавление задачи:

insert into task (id, period, next_run_time) values ($1, $2, $3);

Получение следующей задачи:

with next_task as (
    select id from task
    where status = 0
      and next_run_time <= localtimestamp
    limit 1
    for update skip locked
)
update task
set
    status = 1
from next_task
where task.id = next_task.id
returning task.id;

Завершение задачи и планирование следующего запуска:

update task
set
    status = 0,
    next_run_time = next_run_time + make_interval(secs => period)
where id = $1

Вместо заключения

В создании специализированной очереди задач средствами РСУБД нет ничего сложного.

"Самопальная" очередь будет отвечать даже самым диким практически любым требованиям бизнеса/предметной области.

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

Автор: yakov-bakhmatov

Источник ^[1]