⚡️Очередь задач на реляционной БД

Зачастую возникает необходимость в рамках приложения сделать очередь задач с такими вводными:

- Не очень высокая нагрузка (меньше 1000 задач в секунду)
- Не хочется тащить в проект Kafka, RabbitMQ, etc.
- Мы не хотим, чтобы задачу одновременно исполняли несколько потоков, нужен какой-то механизм синхронизации

Посмотрим на нашу модель данных:

create type task_status as enum ('NEW', 'COMPLETED', 'FAILED');

create table tasks
(
    id           bigserial   not null primary key,
    status       task_status not null,
    scheduled_ts timestamp   not null,

    body         jsonb       not null
);

Если просто делать

select * from tasks where status = 'NEW' and scheduled_ts < now()

То это приведет к ситуации, что одну и ту же задачу могут взять несколько воркеров.

Поэтому используем такой механизм как select for update. Он позволяет выбрать записи, удоволетворяющие условию и сразу их заблокировать до окончания текущей транзакции.

Итого цикл обработки одной задачи будет выглядеть так:

1) Открываем транзакцию
2) Достаем одну никем не заблокированную задачу, которая стоит первой в очереди:

select * from tasks
where status = 'NEW' 
and scheduled_ts < now()
order by scheduled_ts
limit 1
for update skip locked

3) Обрабатываем задачу
4) Выставляем нужный статус
5) Закрываем транзакцию

В приложении этот цикл фактически в while(true) { ... } параллельно исполняют несколько потоков, разгребая очередь.

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

⚡️Transactional Messaging

❓Проблема:

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

Если делать так

begin;
<update statement>;
sendToQueue(event);
commit;

Возможна ситуация, что событие отправится, но транзакция в итоге откатится, например, из-за проблем с IO.

✅ Решение:

В транзакции, в которой происходит изменение в БД будем не отправлять эвент в очередь, а сохранять в базу. Таким образом, в силу атомарности транзакций нам уже СУБД гарантирует, что эвент сохранился <=> произошел апдейт.

И наконец, чтобы доставить сообщение до очереди можно использовать идею очереди на реляционной БД, которая была описана в предыдущем посте: https://news.1rj.ru/str/MicroservicesQuestions/5

Итоговая версия:

begin;
<update statement>
insert into tasks … # спустя какое-то время доставится до очереди
commit;

⚡️OLTP vs OLAP

1️⃣ OLTP (Online Transactional Processing) - система обработки данных, при которой происходят частые короткие транзакции, преимущественно на запись.

- В качестве запросов используются простые insert, update, delete
- Время ответа: миллисекунды или меньше
- Объем БД обычно не очень большой
- Примеры СУБД: PostgreSQL, MongoDB, MySQL

2️⃣ OLAP (Online Analytical Processing) - система обработки данных, суть которой заключается в долгих читающих аналитических запросах.

- В качестве запросов используются сложные select запросы, обычно с join-ами, агрегациями и выборками большого числа строк
- Время ответа: до нескольких дней
- Объем БД может быть очень большим
- Примеры СУБД: Apache HBase, Clickhouse, Elasticsearch

Если кому-то интересна Kotlin разработка, то welcome на мой второй канал с аналогичным контентом, но уже по Kotlin 🙂