Исходный код вики Примеры
Версия 8.30 от Alexandr Fokin на 2026/07/27 03:59
Скрыть последних авторов
| 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.25 | 68 | |((( |
| 69 | Проблема: изменение счетчика не привязано к основной транзакции БД. | ||
| 70 | |||
| 71 | Возможно: | ||
| 72 | |||
| 73 | 1. До коммита транзакции уменьшаем счетчик и записываем id дочернего процесса в redis hashset. Если счетчик 0, то публикуем triggerEvent через TransactionalOutbox. | ||
| |
8.26 | 74 | (На старте транзакции проверить и удалить из HashSet и если удалили, то увеличить счетчик на 1 - значит была упавшая транзакция). |
| |
8.25 | 75 | 1. После коммита транзакции удаляем id дочернего процесса из redis hashset. |
| |
8.28 | 76 | (Таким устраняем возможность потери сигнала, а также дублирование прокрутки счетчика). |
| |
8.25 | 77 | 1. Trigger при получении события смотрит значение счетчика и HashSet. |
| |
8.30 | 78 | Если счетчик 0 и HashSet пустой, то активируем триггер. |
| 79 | Внутри хендлера проверяем что все процессы завершены (если нет, то значит были проблемы переходим в аварийный режим и ждем сигналов от незавершенных процессов). | ||
| |
8.25 | 80 | Иначе если HashSet не пустой, то можно попытаться проверить статус процесса (если он завершен, то можно удалить значение из HashSet). |
| |
8.17 | 81 | ))) |
| |
8.25 | 82 | |Проблемы: Потеря счетчика целиком или искажения значения в случае падения Redis. |
| |
8.30 | 83 | В таком случае дочерние процессы переходят в аварийный режим (каждый раз публикуют событие сигнал, а также событие взведения таймера чтобы сгладить нагрузку). |
| |
8.25 | 84 | ))) |
| 85 | ))) | ||
| |
8.5 | 86 | |2|(% style="width:188px" %)Transaction outbox stream process.|(% style="width:1268px" %)[[image:TransactionOutbox. Sequence.jpg]] |
| 87 | |3|(% style="width:188px" %)Stream trigger|(% style="width:1268px" %)((( | ||
| |
1.2 | 88 | | |((( |
| 89 | * Позволяет убрать лишние запросы пробуждения процесса (когда он и так запущен). | ||
| |
8.10 | 90 | * __Позволяет полностью убрать задержку после остановки процесса__ (если есть новое сообщения, то он сразу же будет пробужден). |
| |
1.2 | 91 | За счет того, что триггер точно знает, что есть новые сообщения и процесс только что уснул. |
| 92 | * Вводит 2 типа события, 1 сигнал о новом сообщении (содержит offset значение), 2 - процесс идет спать (содержит offset значение). | ||
| 93 | * Вводит дополнительное состояние в триггер: максимальный offset сообщения, максимальный offset обработанного процессом сообщения, флаг состояния сна процесса. | ||
| |
8.10 | 94 | * В некоторых случаях позволяет не выполнять wakeup код в конце сессии обработки (если отключить wakeup, оставить только stream trigger) |
| 95 | (блокировка и обновление wakeup entity, проверка wakeup условия), __улучшает перформанс такта работы__. | ||
| |
1.2 | 96 | ))) |
| 97 | |Алгоритм триггера.|((( | ||
| 98 | * При получении события о засыпании процесса: | ||
| 99 | Фиксирует смещение процесса обработки и сравнивает со смещением сообщения. | ||
| 100 | Если все сообщения обработаны, то не пробуждает процесс, иначе пробуждает процесс. | ||
| 101 | * При получении события о новом сообщении: | ||
| 102 | Фиксирует новое наибольшее смещение. | ||
| 103 | Если процесс не спит (по флагу в триггере), то ничего не делает. | ||
| 104 | Если процесс спит (по флагу), то пробуждает процесс. | ||
| 105 | |||
| 106 | Отслеживает смещение обработки процесса и последнего события. | ||
| 107 | Ожидает от процесса события о том, что он все обработал, его последнее смещение и он идет спать. | ||
| 108 | Если есть сообщения со смещением больше чем указал процесс, то делает гарантированное пробуждение процесса. | ||
| 109 | Когда поступает сигнал о новом сообщении (от отправителя сообщения), то обновляет данные о максимальном смещении и пробуждает процесс, если он спит | ||
| 110 | ))) | ||
| |
8.9 | 111 | |Заготовка|[[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 | 112 | ))) |
| |
8.5 | 113 | |4|(% style="width:188px" %)Групповое действие|(% style="width:1268px" %)((( |
| |
8.7 | 114 | | |Действие, которое нужно применить к диапазону строк (сравнительно большому), независимо для каждой строки. |
| |
8.6 | 115 | Наличие у строк упорядоченного столбца (для выделения диапазонов). |
| |
8.2 | 116 | | |((( |
| |
8.3 | 117 | |(% style="width:888px" %)Родительские процесс определяет границы диапазона [min, max].|(% style="width:266px" %){{code language="none"}}select min(), max() |
| 118 | where condition(){{/code}} | ||
| 119 | |(% style="width:888px" %)Родительский процесс нарезает диапазон [min, max] на поддиапазоны. На каждый поддиапазон создается дочерний процесс.|(% style="width:266px" %) | ||
| 120 | |(% style="width:888px" %)Каждый дочерний процесс обрабатывает свой поддиапазон строк (параллельно).|(% style="width:266px" %)Внутри поддиапазона может использоваться keyset пагинация. | ||
| |
8.8 | 121 | |(% style="width:888px" %)Родительский процесс ожидает завершения дочерних процессов (см. пример 1).|(% style="width:266px" %) |
| |
8.2 | 122 | ))) |
| 123 | ))) | ||
| |
8.24 | 124 | |
| 125 | ---- | ||
| 126 | |||
| 127 | ==== Внутренние ссылки: ==== | ||
| 128 | |||
| 129 | ====== Дочерние страницы: ====== | ||
| 130 | |||
| 131 | {{children/}} | ||
| 132 | |||
| 133 | ====== Обратные ссылки: ====== | ||
| 134 | |||
| 135 | {{velocity}} | ||
| 136 | #set ($links = $doc.getBacklinks()) | ||
| 137 | #if ($links.size() > 0) | ||
| 138 | #foreach ($docname in $links) | ||
| 139 | #set ($rdoc = $xwiki.getDocument($docname).getTranslatedDocument()) | ||
| 140 | * [[$escapetool.xml($rdoc.fullName)]] | ||
| 141 | #end | ||
| 142 | #else | ||
| 143 | No back links for this page! | ||
| 144 | #end | ||
| 145 | {{/velocity}} | ||
| 146 | |||
| 147 | ---- |