Имеется некоторый пул транзакций для загрузки в учётную систему из внешнего источника. Для простоты пусть это будет таблица apps.t_import , а сама транзакция содержит только номер, дату и сумму. Загрузка транзакций выполняется загрузчиком, который вызовет стандартный public API учётной системы, в виде 10 параллельных сессий, вместе которые обеспечивают загрузку данных со скоростью не менее 200 транзакций в секунду. Исходный пул транзакций постоянно пополняется новыми данными, это могут быть дубликаты или обновления уже имеющихся записей. Напишите проект кода загрузчика, который позволит избежать блокировок, дубликатов и пропущенных транзакций.

Может ли одна и та же транзакция одновременно или со сдвигом по времени грузиться более чем одним параллельным потоком?

Как организовать пул таким образом, чтобы заранее не назначать (не распределять) задания между параллельными сессиями?

Какие нюансы могут возникнуть, если внешних систем, использующих общий пул, будет несколько? Если пулов с параллельным доступом будет несколько?

• Используемая база данных: Oracle 19c EE;
• Для простоты используется одна схема importer, в которой расположены все объекты, участвующие в загрузке, как то:
  • Таблица импорта;
  • API, которому передается транзакция;
  • Задания, имитирующие входящую нагрузку;
  • Задания, обрабатывающие таблицу t_import;
  • Процедурный код, вызываемый заданиями;
• Для простоты под действиями API будет пониматься обновление поля "Обработано" в "Истина" в таблице транзакций;
• Уникальность транзакции определяется по паре "номер - дата"
• Для простоты интерфейсный API состоит из одной функции вставки строки и при желании может быть пополнен процедурами передачи массива транзакций с последующими проверками и вставкой

Для запуска установки требуется выполнить следующее:

cd src
sqlplus privileged_user@db "@deploy.sql"

Например:

sqlplus sys@db as sysdba "@deploy.sql"

При наличии установленного фреймворка utPlSQL можно выполнить установку и запуск тестов после установки основных компонентов:

cd src
sqlplus importer@db "@tests.sql"

Для запуска примера требуется выполнить следующее:

cd src
sqlplus importer@db "@example.sql"

В примере запускается 3 процесса, которые вставляют данные, и 4 процесса-обработчика. Результаты обработки можно проверить следующим запросом:

with
w_results as (
    select process_result,
           count(*) as cnt,
           decode(process_result,
               0, 'Unprocessed',
               1, 'Processed',
               2, 'Failed'
           ) as label
    from t_import
    group by process_result
)
select listagg(
    label || ': ' || lpad(to_char(cnt), 9), ' | '
) within group(order by process_result) as result
from w_results;

Запрос возвращает число записей в разрезе статуса их обработки.

Для обработки таблицы импорта используется 4 обработчика, которые вычитывают задания на базе бизнес ключа. Это обеспечивает "привязку" набора изменений одной транзакции к конкретному обработчику. Кроме того, это означает, что никакая транзакция не может быть обработана другим обработчиком, что не требует блокировки обрабатываемой строки.

Плюсы имплементации:

• отсутствие блокировок на строках таблицы при обработке;
• обработка записей в порядке поступления (поскольку запрос использует составной индекс с входящей в него датой вставки);

Минусы имплементации:

• Нет возможности "на лету" увеличить/уменьшить число потоков, т.к. это затрагивает функцию распределения заданий
• Ввиду захода по одному индексу в интерфейсную таблицу наблюдаются ожидания вида buffer busy waits / index contention, поскольку индекс становится "горячим"

Возможные улучшения:

• Добавление управляющего потока, который бы поднимал обработчики на базе таблицы-словаря;
• Данный управляющий поток мог бы самостоятельно использовать хэш-функцию и вычитывать данные "оптом", чтобы предотвратить "нагревание" индекса;
• Добавление AQ очередей или простых таблиц для передачи обработчикам контекста исполнения;
• Кроме того, можно собирать сразу же полную строку и передавать обработчику все данные для его работы, чтобы уменьшить число заходов в таблицу;
• Можно секционировать таблицу t_import по списку и при вставке новых строк передавать идентификатор секции. Сам идентификатор переключать, например, раз в минуту, чтобы вставка происходила всегда в одну секцию, а обработчики работали с предыдущей секцией, что уменьшит конкуренцию между продюсерами и консьюмерами.

Может ли одна и та же транзакция одновременно или со сдвигом по времени грузиться более чем одним параллельным потоком?

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

Как организовать пул таким образом, чтобы заранее не назначать (не распределять) задания между параллельными сессиями?

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

Какие нюансы могут возникнуть, если внешних систем, использующих общий пул, будет несколько? Если пулов с параллельным доступом будет несколько?

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