Изменения документа Движок cccc1808. ProcessEngine
Редактировал(а) Alexandr Fokin 2026/05/11 11:57
От версии 31.20
отредактировано Alexandr Fokin
на 2026/05/11 11:56
на 2026/05/11 11:56
Изменить комментарий:
К данной версии нет комментариев
К версии 23.1
отредактировано Alexandr Fokin
на 2026/04/10 19:42
на 2026/04/10 19:42
Изменить комментарий:
Удаление картинки Родительский дочерний процесс. Sequence.jpg
Сводка
-
Свойства страницы (1 изменено, 0 добавлено, 0 удалено)
Подробности
- Свойства страницы
-
- Содержимое
-
... ... @@ -1,43 +1,30 @@ 1 -|(% style="width:132px" %) |(% style="width:1301px" %)((( 2 - |Теги поиска|cccc1808. ProcessEngine, cccc1808.ProcessEngine, Process engine1 +|(% style="width:132px" %)Теги поиска|(% style="width:1301px" %)((( 2 +cccc1808. ProcessEngine, cccc1808.ProcessEngine 3 3 Очередь задач, Система обработки процессов, Движок обработки процессов. 4 4 [[Процесс>>doc:Архитектура и модели.Модели.Процесс.WebHome]] 5 -|Описание|Универсальный движок для выполнения процессов и очередей задач, позволяющий комбинировать несколько подходов к обработке (см особенности). 6 -|Термины|((( 7 -* Процесс является единицей исполнения. В реализации может содержать машину состояний. 8 -* Система триггеров используется для таймеров и передачи сигналов для процессов (с оптимизацией нагрузки). 9 9 ))) 10 -|Репозиторий|[[https:~~/~~/github.com/cccc1808/cccc1808.ProcessEngine>>https://github.com/cccc1808/cccc1808.ProcessEngine]] 11 -|Некоторые цели|((( 12 -* Реализация механизмов для эффективной конкурентной передачи сигнала в процесс. 13 -* Реализация механизмов для минимизации задержки передачи сигнала в процесс. 14 -* Возможность управлять параметрами, влияющими на Throughput и Latency. 15 -* Минимизация нагрузки на БД. 16 -* Наличие общей платформы (компонентов и абстракций), которая решает широкий спектр задач за счет конфигурации. 17 -))) 18 -))) 19 -|(% style="width:132px" %)Развертывание|(% style="width:1301px" %)((( 6 +|(% style="width:132px" %) |(% style="width:1301px" %)Универсальный движок для выполнения процессов и очередей задач, позволяющий комбинировать несколько подходов к обработке (см особенности). 7 +|(% style="width:132px" %) |(% style="width:1301px" %)Процесс является единицей исполнения. В реализации может содержать машину состояний. 8 +Система триггеров используется для таймеров и передачи сигналов для процессов (с оптимизацией нагрузки). 9 +|(% style="width:132px" %)Разветывание|(% style="width:1301px" %)((( 20 20 |(% style="width:150px" %)База данных|(% style="width:1177px" %)((( 21 21 Для надежного хранения данных процессов и триггеров. 22 22 23 -* Есть реализация под [[EntityFramework>>doc:Разработка.NET.Работа с БД.EntityFramework | EntityFrameworkCore.WebHome]]. 24 -* Есть реализация под [[linq2db>>doc:Разработка.NET.Работа с БД.linq2db.WebHome]] (cccc1808/feature/linq2db). 25 -* Модель позволяет сделать реализацию под другие решения (за счет наличия IProcessContainer и методов Update). В коде процесса может использоваться как ChangeTracker, так и прямое обновление БД. 13 +* Сейчас есть реализация под EntityFramework. 14 +* Но модель позволяет сделать реализацию под другие решения ([[linq2db>>doc:Разработка.NET.Работа с БД.linq2db.WebHome]] или чистый Ado.Net) (за счет наличия IProcessContainer и методов Update). 26 26 27 27 Для текущей реализации в качестве хранилище может выступать БД, поддерживающая: 28 28 29 -* Транзакции: 30 -** Транзакции. 31 -** Savepoint. 18 +* Транзакции: транзакции. 19 +* Транзакции: savepoint. 32 32 (если используются, можно обрабатывать каждый шаг отдельной транзакцией или весь процесс без savepoint изоляции ошибок между шагами). 33 -* [[Блокировки>>doc:Разработка.Базы данных.SQL.Механизмы.Транзакции и блокировки.WebHome]]: 34 -** updatelock. 35 -** updatelock skip locked. 21 +* Блокировка: updatelock. 22 +* Блокировка: updatelock skip locked. 36 36 (частично можно обойтись без него). 37 -* *sharelock24 +* Блокировка: sharelock 38 38 (можно обойтись без него без сильного влияния) 39 -* [[Уровни изоляции>>doc:Разработка.Базы данных.SQL.Механизмы.Транзакциииблокировки.Уровни изоляции| Isolation levels.WebHome]]: работает на read committed, то что нужно блокируется руками.40 -* Для некоторых кейсов желательно возможность выполнить [[Upsert>>doc:Разработка.Базы данных.SQL.Сценарии и вопросы.Insert orupdate\. Upsert.WebHome]](insert on conflict).26 +* Уровень изоляции: работает на read committed, то что нужно блокируется руками. 27 +* Для некоторых кейсов желательно возможность выполнить upsert (insert on conflict). 41 41 ))) 42 42 |(% style="width:150px" %)Брокер сообщений|(% style="width:1177px" %)((( 43 43 Используется для накопления и доставки TriggerEvent. ... ... @@ -48,8 +48,8 @@ 48 48 ))) 49 49 ))) 50 50 |(% style="width:132px" %)Особенности|(% style="width:1301px" %)((( 51 -|(% style="width:159px" %) Пакетныетранзакции(батчинг).|(% style="width:1168px" %)(((52 -Возможность использовать и комбинировать типы выполнения для разных типов процессов:38 +|(% style="width:159px" %)Батчинг при выполнении.|(% style="width:1168px" %)((( 39 +Возможность использовать и комбинировать разные типы выполнения как 53 53 54 54 * (1 транзакция - 1 процесс), 55 55 * (1 транзакция - N процессов). ... ... @@ -56,8 +56,7 @@ 56 56 ))) 57 57 |(% style="width:159px" %)Изоляция шагов и процессов (внутри оной транзакции)|(% style="width:1168px" %)((( 58 58 * Изоляция через db savepoint. 59 -* Для [[EntityFramework ~| EntityFrameworkCore>>doc:Разработка.NET.Работа с БД.EntityFramework | EntityFrameworkCore.WebHome]] возможность делать InMemory снимок [[Создание снимка ChangeTracker>>doc:Разработка.NET.Работа с БД.EntityFramework | EntityFrameworkCore.Сценарии и проблемы.Создание снимка ChangeTracker.WebHome]] 60 -(может использоваться если все изменения накапливаются InMemory (без прямой записи в БД) или же с указанием ручной компенсации, а также ошибка на записи в БД не велика). 46 +* Для [[EntityFramework ~| EntityFrameworkCore>>doc:Разработка.NET.Работа с БД.EntityFramework | EntityFrameworkCore.WebHome]] возможность делать InMemory снимок [[Создание снимка ChangeTracker>>doc:Разработка.NET.Работа с БД.EntityFramework | EntityFrameworkCore.Сценарии и проблемы.Создание снимка ChangeTracker.WebHome]]. 61 61 * Допустима реализация на основе того, что InMemory состояние процесса (и задействованных бизнес сущностей) реализовано в виде Immutable компонентов. В этом случае достаточно просто сохранять и восстанавливать ссылку. 62 62 Но предполагаю, что в большинстве бизнес моделей не так и это не не будет востребованным. 63 63 ))) ... ... @@ -74,7 +74,6 @@ 74 74 См. пример 2. 75 75 ))) 76 76 |(% style="width:159px" %)Перехват ошибок|(% style="width:1168px" %)Перехват и обработка ошибок, если процесс выкинул exception в движок. Реализацию простого retry с задержкой (создается триггер на следующую попытку). 77 -В случае пакетной транзакции движок не знает какой конкретно из процессов породил ошибку (если она не перехвачена вручную), то ошибка выставляется на все незавершенные процессы. 78 78 |(% style="width:159px" %)Параллельное выполнение|(% style="width:1168px" %)Допускается запуск нескольких раннеров (на разных нодах), работающих с одной таблицей процессов для распределения нагрузки между ними. 79 79 Допускается фильтрация типов процессов между нодами (чтобы нода выполняла только определенные типы процессов, в том числе по приоритету). 80 80 Доступно для раннеров процессов и триггеров. ... ... @@ -90,34 +90,51 @@ 90 90 2) N/2 за вторую сессию выполнения (транзакцию) и уснуть до окончания дочерних процессов. 91 91 1. Процесс - групповое действие (когда нужно применить действие к строкам таблицы по указанному фильтру). 92 92 ))) 93 -|(% style="width:159px" %)Range preload process data|(% style="width:1168px" %)((( 94 -(Необязательно) возможность в начале сессии обработки пакетно предзагружать данные и бизнес сущности для процессов (Range query) из обрабатываемого батча для оптимизации чтения (если используется обработка пакета процессов в одной транзакции). 78 +|(% style="width:159px" %)Range preload process data|(% style="width:1168px" %)(Необязательно) возможность в начале сессии обработки пакетно предзагружать данные и бизнес сущности для процессов (Range query) из обрабатываемого батча для оптимизации чтения (если используется обработка пакета процессов в одной транзакции). 79 +))) 80 +|(% style="width:132px" %)Примеры|(% style="width:1301px" %)((( 81 +|(% style="width:32px" %)1|(% style="width:171px" %)Родительский процесс, N дочерних процессов.|(% style="width:1066px" %)((( 82 +|(% style="width:870px" %)((( 83 +В данном примере имеется в виду, что дочерние процессы могут выполняться параллельно другу и независимо друг от друга, но в конце должны оповестить родительский процесс о необходимости продолжения обработки. 84 +Если речь идет о каких-либо зависимостях порядка выполнения в дочерних процессах, то это может контролировать дочерний процесс (выделяя группу, которую сейчас можно запустить и ожидая окончания). 85 +))) 86 +|(% style="width:870px" %)((( 87 +1. Родительский процесс создает триггер со счетчиком N, создает и запускает дочерние процессы, засыпает. 88 +1. Дочерние процесс при завершении публикует TriggerEvent. 89 +1. TriggerConsumerRunner периодически считывает батч TriggerEvent, уменьшает считчик триггера и делает запись в БД. За счет агрегации событий завершения процессов мы уменьшаем нагрузку на БД. 90 +1. Когда все дочерние процессы отработали TriggerConsumerRunner получает значение счетчика 0 и взводит триггер. 91 +1. Триггер пробуждает родительский процесс для дальнейшего выполнения. 92 +))) 93 +|(% style="width:870px" %)TriggerEvent публикуются без использования TransactionOutbox напрямую в брокер после коммита транзакции (иначе мы бы нагружали БД). 94 +|(% style="width:870px" %)((( 95 +Предполагаем, что основную часть времени система работает стабильно, но допускается ситуация, что транзакция закоммитилась, но TriggerEvent не смогли опубликоваться (остановка сервиса без graceful shutdown, проблемы соединения или работы с брокером сообщений). 95 95 96 - Такжеестьвозможностьиспользоватькак дополнительное ограничениепонекоторойоценкеприпакетнойобработке.97 - (Напримермызагрузили10процессов,у каждогопо50 необработанныхсообщений,мыотбираем только limit400сообщенийи связанные процессыдлятекущейтранзакции).97 +Для таких случаев создается страхующий триггер (1 общий на тип процесса). Этот триггер запускается периодически и проходится по всем ожидающим процессам, проверяя условие (в реализации можно использовать keyset пагинацию) (в реализации можно использовать join для проверки условия). 98 +Этот триггер выполняется периодически с более крупной временной задержкой. В случае обнаружения потери TriggerEvent, он поднимет заклинивший родительский процесс и он будет обработан (но позже). Можно установить этому триггеру низкий приоритет. 98 98 ))) 100 +|(% style="width:870px" %)[[image:Родительский дочерний процесс. Sequence.jpg]] 99 99 ))) 102 +|(% style="width:32px" %)2|(% style="width:171px" %)Transaction outbox stream process.|(% style="width:1066px" %)[[image:TransactionOutbox. Sequence.jpg]] 103 +|(% style="width:32px" %)3|(% style="width:171px" %)Stream trigger|(% style="width:1066px" %)((( 104 +|(% style="width:94px" %) |(% style="width:1002px" %)Позволяет убрать лишние запросы пробуждения процесса (когда он и так запущен). 105 +Вводит 2 типа события, 1 сигнал о новом сообщении (содержит offset значение), 2 - процесс идет спать (содержит offset значение). 106 +|(% style="width:94px" %)Алгоритм триггера.|(% style="width:1002px" %)((( 107 +* При получении события о засыпании процесса: 108 +Фиксирует смещение процесса обработки и сравнивает со смещением сообщения. 109 +Если все сообщения обработаны, то не пробуждает процесс, иначе пробуждает процесс. 110 +* При получении события о новом сообщении: 111 +Фиксирует новое наибольшее смещение. 112 +Если процесс не спит (по флагу в триггере), то ничего не делает. 113 +Если процесс спит (по флагу), то пробуждает процесс. 100 100 101 ----- 102 102 103 -==== Внутренние ссылки: ==== 116 +Отслеживает смещение обработки процесса и последнего события. 117 +Ожидает от процесса события о том, что он все обработал, его последнее смещение и он идет спать. 118 +Если есть сообщения со смещением больше чем указал процесс, то делает гарантированное пробуждение процесса. 119 +Когда поступает сигнал о новом сообщении (от отправителя сообщения), то обновляет данные о максимальном смещении и пробуждает процесс, если он спит 120 +))) 121 +|(% style="width:94px" %) |(% style="width:1002px" %)TODO: 122 +))) 123 +))) 104 104 105 -====== Дочерние страницы: ====== 106 - 107 -{{children/}} 108 - 109 -====== Обратные ссылки: ====== 110 - 111 -{{velocity}} 112 -#set ($links = $doc.getBacklinks()) 113 -#if ($links.size() > 0) 114 - #foreach ($docname in $links) 115 - #set ($rdoc = $xwiki.getDocument($docname).getTranslatedDocument()) 116 - * [[$escapetool.xml($rdoc.fullName)]] 117 - #end 118 -#else 119 - No back links for this page! 120 -#end 121 -{{/velocity}} 122 - 123 ----- 125 +