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

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

- Не очень высокая нагрузка (меньше 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 🙂

⚡️2PC (Two Phase Commit)

❓Проблема:

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

✅ Решение:

Разделим нашу распределенную транзакцию на несколько этапов:

1. Просим все базы подготовить транзакции к коммиту. После этого этапа пока никаких изменений в базах не произошло.

❗️Важно, что СУБД должна поддерживать эту возможность, и что подготовка транзакции должна гарантировать успешный коммит в будущем.

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

❗️На втором этапе также могут быть проблемы: например, после коммита транзакции в первой базе, вторая база стала недоступна. Самым простым решением здесь будет пытаться ретраить запрос, но есть и другие политики обработки подобных ситуаций.

⚡️Почему большинство баз данных на самом деле не CP и не AP?

Крутая статья от Мартина Клеппмана (автор той самой книги с кабанчиком) на тему того, что на самом деле означают Consistency, Availability, Partition tolerance из CAP-теоремы, и почему большинство популярных БД удоволетворяют лишь Partition tolerance.

⚡Distributed Locking в Redis

❓Проблема:

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

✅ Решение:

Используем Redis в качестве механизма синхронизации.

1. У нас один Redis мастер. В этом случае все просто:

1) Создаем запись, если таковая не существует по заданному ключу (NX) с таймаутом в 30000 миллисекунд (PX)

SET resource_name random_value NX PX 30000

2) Далее после использования ресурса удаляем лок, если он еще принадлежит нам - именно для этого используется random_value при блокировании.

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

2. У нас несколько Redis мастеров. Используем алгоритм Redlock, основная идея которого в следующем:

1) Получаем текущее время
2) Пытаемся сходить во все узлы и захватить там лок
3) Считаем время, которое потребовалось на захваты локов.
4) Если лок успешно захвачен на большинстве (n/2 + 1) узлов и время захвата меньше времени таймаута лока - то считаем, что лок успешно захвачен. В противном случае шлем запросы на разблокировку во все узлы.

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

❗️Во втором алгоритме считается, что часы между узлами синхронизированы.

❗️Если локи нужны для обеспечения консистентности данных, то будьте очень осторожны с этими алгоритмами, в противном случае есть риск получать ситуацию, когда два потока одновременно используют ресурс.

⚡Идемпотентность API

❓Проблема:

Иногда возникает потребность, что несколько вызовов метода API должны иметь тот же эффект, что и один вызов. Например, чтобы в случае ретраев из-за сетевых проблем POST запрос не создавал снова ту же сущность.

✅ Решение:

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

❗️Существуют различные реализации механизма проверки, что запрос таким ключом уже обрабатывался. Но всегда стоит помнить, что наивная реализация check-then-act без каких-либо блокировок подвержена race-conditions (оба клиента почти одновременно проверили, что заданного ключа нет, и оба пошли исполнять запрос).

Имхо, наиболее простая реализация, когда хранилищем выступает реляционная БД с одним шардом, выглядит так:

Создадим таблицу для хранения уже обработанных ключей:

create table used_keys (
    value varchar(256) not null primary key
);

И теперь обработка нашего запроса выглядит примерно так:

begin transaction;
— проверка
insert into used_keys (value)
values ('some_key')
on conflict do nothing
returning *;
— если запрос ничего не вернул, значит уже обрабатывали
— если вернул, то обрабатываем запрос
commit;

insert … on conflict do nothing в рамках транзакции блокирует вставку с данным ключом, пока текущая транзакция либо не закомитится, либо не роллбекнется, что решает проблему с check-then-act: второй клиент просто будет ждать, пока первый успешно/неуспешно обработает запрос.

⚡Декомпозиция системы на модули/микросервисы

Хороший доклад на тему того, как разбить систему так, чтобы она не превратилась в Big ball of mud, и чтобы свести к минимуму синхронизацию между различными сервисами.

⚡Session and Token Authentication

❓Проблема:

Хотим аутентифицировать пользователя и хранить какую-то сессионную информацию, привязанную к пользователю.

✅ Решение:

1. Session based Authentication

Принцип работы обычно выглядит так:
- Пользователь отправляет POST запрос /login с телом {username, password}
- Сервер сохраняет в БД сессию и отдает sessionId пользователю в куки
- При повторных обращениях клиент передает sessionId, а сервер по нему аутентифицирует простым лукапом в базе

В этом случае пользовательская сессия хранится в базе данных на сервере.

2. Token based Authentication

Принцип работы обычно выглядит так:
- Пользователь отправляет POST запрос /login с телом {username, password}
- Сервер отдает Json Web Token, клиент у себя его локально сохраняет
- При повторных обращениях JWT передается в хедере запроса, сервер его валидирует и достает данные пользователя

В этом случае токен хранится в локальном хранилище на клиенте.

❗️Если у вас ожидается много пользователей и высокая нагрузка, то вероятно стоит сразу задуматься о JWT. Поскольку для Session based Authentication требуется где-то на сервере хранить эти сессии, валидация требует обращения к БД. Если хранилище шардированное, нужно контролировать, чтобы запросы по одной и той же сессии приходили на один и тот же шард. В свою очередь, JWT позволяет производить stateless валидацию и не хранить сессионные данные на сервере.

⚡️Согласованность данных в межсервисных транзакциях

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

⚡Sticky sessions

❓Проблема:

Есть один инстанс приложения, использующего веб-сокеты, например, мессенджера. Хотим добиться горизонтального масштабирования, добавив несколько инстансов приложения + балансировщик нагрузки.

✅ Решение:

Использовать Sticky session load balancing, который заключается в том, что запросы от одного клиента будут приходить на один и тот же сервер. В контексте веб-сокетов это особенно важно, поскольку часто требуется, чтобы запросы от клиента приходили на тот же бэкенд, с которым изначально было установлено соединение. Например, это полезно, если у нас есть локальный кеш, в котором хранятся клиентские данные, и мы хотим чтобы после разрыва соединения, следующий запрос на установку соединения пришел на тот же бэкенд.

Принцип работы следующий: балансировщик нагрузки смотрит на какие-то параметры запроса, которые будут постоянны для одного клиента/одной сессии и по ним определяет, на какой сервер переадресовать запрос. Например, можно использовать хеш ip адреса клиента.

⚡Паттерны обработки ошибок в Apache Kafka

❓Проблема:

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

✅ Решение:

Ответ зависит от специфики приложения: проигнорировать это сообщение, остановить чтение, перебросить в error-топик и тд. В статье описываются популярные паттерны обработки ошибок в Apache Kafka вместе с вариантами использования.