Исходный код вики Примеры
Версия 8.34 от Alexandr Fokin на 2026/07/27 20:22
Скрыть последних авторов
| author | version | line-number | content |
|---|---|---|---|
 |
8.5 | 1 | |1|(% style="width:188px" %)1 родительский процесс и N дочерних процессов.|(% style="width:1268px" %)((( |
| |
8.4 | 2 | |В данном примере имеется в виду, что дочерние процессы могут выполняться параллельно другу и независимо друг от друга, но в конце должны оповестить родительский процесс о необходимости продолжения обработки. |
| 3 | Если речь идет о каких-либо зависимостях порядка выполнения в дочерних процессах, то это может контролировать дочерний процесс (выделяя группу, которую сейчас можно запустить и ожидая окончания). | ||
| |
1.2 | 4 | |((( |
| |
8.4 | 5 | |((( |
| 6 | Вариант 1: CounterTrigger. | ||
| |
1.2 | 7 | ))) |
| 8 | |((( | ||
| 9 | 1. Родительский процесс создает триггер со счетчиком N, создает и запускает дочерние процессы, засыпает. | ||
| 10 | 1. Дочерние процесс при завершении публикует TriggerEvent. | ||
| 11 | 1. TriggerConsumerRunner периодически считывает батч TriggerEvent, уменьшает считчик триггера и делает запись в БД. За счет агрегации событий завершения процессов мы уменьшаем нагрузку на БД. | ||
| 12 | 1. Когда все дочерние процессы отработали TriggerConsumerRunner получает значение счетчика 0 и взводит триггер. | ||
| 13 | 1. Триггер пробуждает родительский процесс для дальнейшего выполнения. | ||
| 14 | ))) | ||
| 15 | |TriggerEvent публикуются без использования TransactionOutbox напрямую в брокер после коммита транзакции (иначе мы бы нагружали БД). | ||
| 16 | |((( | ||
| 17 | Предполагаем, что основную часть времени система работает стабильно, но допускается ситуация, что транзакция закоммитилась, но TriggerEvent не смогли опубликоваться (остановка сервиса без graceful shutdown, проблемы соединения или работы с брокером сообщений). | ||
| 18 | |||
| 19 | Для таких случаев создается страхующий триггер (1 общий на тип процесса). Этот триггер запускается периодически и проходится по всем ожидающим процессам, проверяя условие (в реализации можно использовать keyset пагинацию) (в реализации можно использовать join для проверки условия). | ||
| 20 | Этот триггер выполняется периодически с более крупной временной задержкой. В случае обнаружения потери TriggerEvent, он поднимет заклинивший родительский процесс и он будет обработан (но позже). Можно установить этому триггеру низкий приоритет. | ||
| 21 | ))) | ||
| |
8.1 | 22 | |[[image:Родительский дочерний процесс. Sequence.jpg]] |
| |
1.2 | 23 | ))) |
| |
8.4 | 24 | |((( |
| |
8.14 | 25 | Вариант №2: |
| |
8.4 | 26 | |
| |
8.17 | 27 | Мы просто ставим TimerTrigger на условно 1-5-10 минут (насколько важна задержка) и перепроверяем условие завершения. |
| |
8.4 | 28 | В этом случае будет |
| 29 | |||
| |
8.17 | 30 | * Из минус: что родительский процесс узнает о завершении дочерних процессов с задержкой. |
| 31 | Если дочерний процесс падает в ошибку, TimerTrigger все равно будет крутиться и создавать пустую нагрузку. | ||
| |
8.22 | 32 | * Из плюсов: будет меньше пишущей нагрузки на БД чем в варианте 1 (но больше читающей - на проверку) т.к. у нас не будет CounterTrigger, но будет периодический запрос на проверку завершения всех дочерних процессов (аналогично страхующему триггер). |
| |
8.17 | 33 | * [Расширенный]: Дочерние процессы в блоке wakeup condition проверяют наличие незавершенных процессов. |
| 34 | ** Если все процессы завершены или (незавершенных процессов мало и нет процессов с ошибкой), то можно опубликовать событие на TimerTrigger на сброс или установку минимальной задержки. | ||
| |
8.4 | 35 | ))) |
| |
8.14 | 36 | |((( |
| 37 | Вариант №3: | ||
| 38 | |||
| 39 | Дочерние процессы выполняются через родительский (ограничение в рамках одной ноды). | ||
| 40 | Точкой выполнения является родительский процесс, который внутри себя (параллельно или последовательно) выполняет дочерние процессы. | ||
| 41 | За счет такого способа у нас также отсутствует конкуренция передачи сигнала в родительский процесс. | ||
| 42 | Но мы ограничены выполнением дочерних процессов одной одной сервиса. | ||
| |
8.15 | 43 | Сложнее контролировать распределение нагрузки, если будет вложенный параллелизм. |
| |
8.16 | 44 | Также решает проблему, если дочерний процесс содержит ожидание (например асинхронный запрос-ответ), тут будет конкуренция сигнала от хендлера ответа к родительскому процессу. |
| |
8.4 | 45 | ))) |
| |
8.17 | 46 | |((( |
| |
8.25 | 47 | Вариант №4: |
| |
8.17 | 48 | |
| 49 | SimpleStreamTrigger + Timer (один из лучших вариантов). | ||
| 50 | |||
| 51 | * Триггер проверяет условие завершения всех дочерних процессов (можно прикинуть количество незавершенных дочерних процессов). | ||
| 52 | ** Если все обработано, то пробуждает процесс и деактивируется. | ||
| |
8.19 | 53 | ** Иначе: |
| 54 | *** деактивируется (до поступления хотя бы одного сигнала), | ||
| 55 | *** взводит признак стрима - процесс ожидает, | ||
| 56 | *** взводит флаг новых сигналов на 0, | ||
| 57 | *** выставляет задержку от оценки количества необработанных процессов (< N - малая задержка, иначе большая задержка). | ||
| |
8.17 | 58 | * [Расширенный]: Дочерние процессы в блоке wakeup condition проверяют наличие незавершенных процессов. |
| |
8.18 | 59 | ** Если все процессы завершены или (незавершенных процессов мало и нет процессов с ошибкой), то можно опубликовать событие на SimpleStreamTrigger на сброс или установку минимальной задержки (в дополнение к сигналу). |
| |
8.21 | 60 | * Читающей нагрузки будет немного больше чем в варианте 2 (чтение триггера на поступлении сигнала), |
| 61 | но пишущей нагрузки будет меньше чем в варианте 1 (запись - только на активации новым сигналом). | ||
| 62 | * Если сигналов нет, то нет пустых срабатываний в отличие от варианта 2 (т.к. нет поступления сигнала от дочерних процессов). | ||
| |
8.14 | 63 | ))) |
| |
8.25 | 64 | |((( |
| |
8.29 | 65 | |Вариант №5: Trigger + Счетчик в [[Redis>>doc:Разработка.Базы данных.NoSQL.Ключ-значение структура.Redis.WebHome]]. |
| |
8.27 | 66 | |InMemory счетчик, нагрузка на БД и конкуренция. |
| 67 | Дочерний процесс уменьшает счетчик. И публикует событие только если счетчик достиг 0. | ||
| |
8.33 | 68 | Совмещает преимущества из варианта 1.1 (при этом не нагружает БД), в случае ошибки переключается на режим 1.4. |
| |
8.25 | 69 | |((( |
| 70 | Проблема: изменение счетчика не привязано к основной транзакции БД. | ||
| 71 | |||
| 72 | Возможно: | ||
| 73 | |||
| |
8.34 | 74 | 1. В начале транзакции (или шага) атомарно проверяем MemberSet, если есть запись то удаляем и увеличиваем счетчик на 1 (означает что процесс уже увеличивал счетчик, но потом было падение). |
| 75 | 1. До коммита транзакции атомарно добавляем значение в MemberSet и уменьшаем счетчик на 1. Если счетчик равен 0, то публикуем TriggerEvent (в ручную компенсацию добавляем вызов из пункта 1 (при откате изоляции шага и компенсации транзакции)). | ||
| |
8.32 | 76 | 1. После коммита транзакции удаляем запись из MemberSet. |
| 77 | 1. Trigger получает событие. (Необязательно) в хендлере может првоерить значения счетчиков и MemberSet. | ||
| |
8.17 | 78 | ))) |
| |
8.32 | 79 | |((( |
| 80 | В случае обнаружения повреждения обработка фактически переходит в режим 1.4: начинает публиковать событий каждый раз и используется задержка. | ||
| 81 | |||
| 82 | Примеры проблемы: | ||
| 83 | |||
| 84 | * Падение InMemory хранилища. Предполагается режим без снимков и удаление ключей. | ||
| 85 | Обнаружение (со стороны дочернего процесса) через отсутствие ключей (проверяется в транзакции). | ||
| 86 | * Дублирование обновления счетчика. | ||
| 87 | Обнаружение (со стороны дочернего процесса) через значение счетчика < 0. | ||
| 88 | Обнаружение (со стороны триггера) через активацию триггера (поступления сигнала от процесса), при этом обнаруживается что не все процессы завершены. | ||
| |
8.25 | 89 | ))) |
| 90 | ))) | ||
| |
8.32 | 91 | ))) |
| |
8.5 | 92 | |2|(% style="width:188px" %)Transaction outbox stream process.|(% style="width:1268px" %)[[image:TransactionOutbox. Sequence.jpg]] |
| 93 | |3|(% style="width:188px" %)Stream trigger|(% style="width:1268px" %)((( | ||
| |
1.2 | 94 | | |((( |
| 95 | * Позволяет убрать лишние запросы пробуждения процесса (когда он и так запущен). | ||
| |
8.10 | 96 | * __Позволяет полностью убрать задержку после остановки процесса__ (если есть новое сообщения, то он сразу же будет пробужден). |
| |
1.2 | 97 | За счет того, что триггер точно знает, что есть новые сообщения и процесс только что уснул. |
| 98 | * Вводит 2 типа события, 1 сигнал о новом сообщении (содержит offset значение), 2 - процесс идет спать (содержит offset значение). | ||
| 99 | * Вводит дополнительное состояние в триггер: максимальный offset сообщения, максимальный offset обработанного процессом сообщения, флаг состояния сна процесса. | ||
| |
8.10 | 100 | * В некоторых случаях позволяет не выполнять wakeup код в конце сессии обработки (если отключить wakeup, оставить только stream trigger) |
| 101 | (блокировка и обновление wakeup entity, проверка wakeup условия), __улучшает перформанс такта работы__. | ||
| |
1.2 | 102 | ))) |
| 103 | |Алгоритм триггера.|((( | ||
| 104 | * При получении события о засыпании процесса: | ||
| 105 | Фиксирует смещение процесса обработки и сравнивает со смещением сообщения. | ||
| 106 | Если все сообщения обработаны, то не пробуждает процесс, иначе пробуждает процесс. | ||
| 107 | * При получении события о новом сообщении: | ||
| 108 | Фиксирует новое наибольшее смещение. | ||
| 109 | Если процесс не спит (по флагу в триггере), то ничего не делает. | ||
| 110 | Если процесс спит (по флагу), то пробуждает процесс. | ||
| 111 | |||
| 112 | Отслеживает смещение обработки процесса и последнего события. | ||
| 113 | Ожидает от процесса события о том, что он все обработал, его последнее смещение и он идет спать. | ||
| 114 | Если есть сообщения со смещением больше чем указал процесс, то делает гарантированное пробуждение процесса. | ||
| 115 | Когда поступает сигнал о новом сообщении (от отправителя сообщения), то обновляет данные о максимальном смещении и пробуждает процесс, если он спит | ||
| 116 | ))) | ||
| |
8.9 | 117 | |Заготовка|[[https:~~/~~/github.com/cccc1808/cccc1808.ProcessEngine/tree/cccc1808/feature/trigger_stream_trigger>>https://github.com/cccc1808/cccc1808.ProcessEngine/tree/cccc1808/feature/trigger_stream_trigger]] |
| |
1.2 | 118 | ))) |
| |
8.5 | 119 | |4|(% style="width:188px" %)Групповое действие|(% style="width:1268px" %)((( |
| |
8.7 | 120 | | |Действие, которое нужно применить к диапазону строк (сравнительно большому), независимо для каждой строки. |
| |
8.6 | 121 | Наличие у строк упорядоченного столбца (для выделения диапазонов). |
| |
8.2 | 122 | | |((( |
| |
8.3 | 123 | |(% style="width:888px" %)Родительские процесс определяет границы диапазона [min, max].|(% style="width:266px" %){{code language="none"}}select min(), max() |
| 124 | where condition(){{/code}} | ||
| 125 | |(% style="width:888px" %)Родительский процесс нарезает диапазон [min, max] на поддиапазоны. На каждый поддиапазон создается дочерний процесс.|(% style="width:266px" %) | ||
| 126 | |(% style="width:888px" %)Каждый дочерний процесс обрабатывает свой поддиапазон строк (параллельно).|(% style="width:266px" %)Внутри поддиапазона может использоваться keyset пагинация. | ||
| |
8.8 | 127 | |(% style="width:888px" %)Родительский процесс ожидает завершения дочерних процессов (см. пример 1).|(% style="width:266px" %) |
| |
8.2 | 128 | ))) |
| 129 | ))) | ||
| |
8.24 | 130 | |
| 131 | ---- | ||
| 132 | |||
| 133 | ==== Внутренние ссылки: ==== | ||
| 134 | |||
| 135 | ====== Дочерние страницы: ====== | ||
| 136 | |||
| 137 | {{children/}} | ||
| 138 | |||
| 139 | ====== Обратные ссылки: ====== | ||
| 140 | |||
| 141 | {{velocity}} | ||
| 142 | #set ($links = $doc.getBacklinks()) | ||
| 143 | #if ($links.size() > 0) | ||
| 144 | #foreach ($docname in $links) | ||
| 145 | #set ($rdoc = $xwiki.getDocument($docname).getTranslatedDocument()) | ||
| 146 | * [[$escapetool.xml($rdoc.fullName)]] | ||
| 147 | #end | ||
| 148 | #else | ||
| 149 | No back links for this page! | ||
| 150 | #end | ||
| 151 | {{/velocity}} | ||
| 152 | |||
| 153 | ---- |