Статьи

Mule Batch Jobs и переменные

Август 18, 2017

1.0 Обзор (случай с переменными потока)

В Муле я наткнулся на экзаменационный вопрос, касающийся взаимосвязи переменных потока и пакетных заданий. Вам нужно обратить внимание на подобные вопросы, если вы хотите сдать сертификационные экзамены Mule. Когда я пытаюсь найти ответ в Google, там не было написано ни одной статьи. Ни в документации по Муле, ни в Интернете. Итак, продолжайте читать и получите представление о том, как каждый тип переменных Mule ведет себя в контексте пакетного задания (конструкция Mule).

Вопрос примерно такой: с учетом следующей пакетной работы,

Рисунок 1.0a.

И учитывая следующую входную полезную нагрузку, что будет flowVar.aCounter в конце при завершении?

[
  { 
    "ID":"001",
    "OriginNum":"1"
  },
  {
    "ID":"002",
    "OriginNum":"2"  
  }
]

Если в начале пакетного потока flowVars.aCounter начинается с единицы (1), то теоретически, у flowVars.aCounter будет всего 5 в конце фазы «На завершении» пакетного задания, верно? Неправильно! Это понятие совершенно неверно! Ниже приведена конфигурация XML Mule пакетного задания. Вот как будет выглядеть ваш мыслительный процесс: полезная нагрузка имеет массив из двух элементов, что означает, что один экземпляр пакетного задания будет запущен, и оба batch_step_1 и batch_step_2 будут выполнены два раза (по одному разу для каждой записи).

    <batch:job name="batchjob_withnoInput">
        <batch:input>
            <logger message="Batch Job Input: #[payload]" level="INFO" doc:name="Logger"/>
            <set-variable variableName="aCounter" value="#[1]" doc:name="flowVars.aCounter = 0"/>
        </batch:input>
        <batch:process-records>
            <batch:step name="Batch_Step_1">
                <scripting:component doc:name="flowVars.aCounter add 1">
                    <scripting:script engine="Groovy"><![CDATA[flowVars.aCounter = flowVars.aCounter + 1]]></scripting:script>
                </scripting:component>
                <logger message="Batch Step 1 flowVars.aCounter: #[flowVars.aCounter]" level="INFO" doc:name="Logger"/>
            </batch:step>
            <batch:step name="Batch_Step_2" >
                <scripting:component doc:name="flowVars.aCounter add 1">
                    <scripting:script engine="Groovy"><![CDATA[flowVars.aCounter = flowVars.aCounter + 1]]></scripting:script>
                </scripting:component>
                <logger message="Batch Step 2 flowVars.aCounter: #[flowVars.aCounter]" level="INFO" doc:name="Logger"/>
            </batch:step>
        </batch:process-records>
        <batch:on-complete>
            <logger message="oncomplete flowVars.aCounter : #[flowVars.aCounter]" level="INFO" doc:name="Logger"/>
        </batch:on-complete>
    </batch:job>

Пакетное задание не является потоком, но оно может ссылаться и выполнять поток. Каждый шаг в пакетном задании выполняется как независимый поток, поэтому область действия переменной потока действительно ограничена отдельными шагами, после того как выполнение покидает пакетный шаг, FlowVars исчезает.

2.0 Реальный ответ

Ниже приведены журналы, собранные, если вы запустите пакетное задание на рисунке 1.0a:

[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Starting input phase
[[batchplayground].HTTP_Listener_Configuration.worker.01] org.mule.api.processor.LoggerMessageProcessor: Batch Job Input: [{OriginNum=1, ID=001}]
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Input phase completed
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.queue.BatchQueueLoader: Starting loading phase for instance '67b3a840-974e-11e7-956a-b62d20524153' of job 'batchjob_withnoInput'
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.queue.BatchQueueLoader: Finished loading phase for instance 67b3a840-974e-11e7-956a-b62d20524153 of job batchjob_withnoInput. 1 records were loaded
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Started execution of instance '67b3a840-974e-11e7-956a-b62d20524153' of job 'batchjob_withnoInput'
[batch-job-batchjob_withnoInput-work-manager.01] org.mule.api.processor.LoggerMessageProcessor: Batch Step 1 flowVars.aCounter: 2
[batch-job-batchjob_withnoInput-work-manager.01] com.mulesoft.module.batch.DefaultBatchStep: Step Batch_Step_1 finished processing all records for instance 67b3a840-974e-11e7-956a-b62d20524153 of job batchjob_withnoInput
[batch-job-batchjob_withnoInput-work-manager.03] org.mule.api.processor.LoggerMessageProcessor: Batch Step 2 flowVars.aCounter: 2
[batch-job-batchjob_withnoInput-work-manager.03] com.mulesoft.module.batch.engine.DefaultBatchEngine: Starting execution of onComplete phase for instance 67b3a840-974e-11e7-956a-b62d20524153 of job batchjob_withnoInput
[batch-job-batchjob_withnoInput-work-manager.03] org.mule.api.processor.LoggerMessageProcessor: oncomplete flowVars.aCounter : 1
[batch-job-batchjob_withnoInput-work-manager.03] com.mulesoft.module.batch.engine.DefaultBatchEngine: Finished execution of onComplete phase for instance 67b3a840-974e-11e7-956a-b62d20524153 of job batchjob_withnoInput
[batch-job-batchjob_withnoInput-work-manager.03] com.mulesoft.module.batch.engine.DefaultBatchEngine: Finished execution for instance '67b3a840-974e-11e7-956a-b62d20524153' of job 'batchjob_withnoInput'. 
                                                                                                      Total Records processed: 1. Successful records: 1. Failed Records: 0
[batch-job-batchjob_withnoInput-work-manager.03] com.mulesoft.module.batch.engine.DefaultBatchEngine: 
[batch-job-batchjob_withnoInput-work-manager.03] com.mulesoft.module.batch.DefaultBatchStep: Step Batch_Step_2 finished processing all records for instance 67b3a840-974e-11e7-956a-b62d20524153 of job batchjob_withnoInput

Так что, если вы когда-нибудь получите такой вопрос во время любого из ваших экзаменов MuleSoft, и если в вопросе вас попросят указать значение переменной потока на этапе «На завершении», то ответ будет равен 1, если переменная потока была инициализирована с помощью 1 (или ноль, если инициализированное значение переменной потока равно нулю). Из журналов видно, что состояние переменной потока flowVars.aCounter не сохраняется при прохождении через этапы пакета, но состояние переменной потока, когда она была впервые инициализирована, переносится на каждый этап и этап пакета , Короче говоря, переменная потока flowVars.aCounter всегда будет начинаться с 1 на каждом шаге пакета, независимо от того, пытается ли шаг пакета увеличить его значение. И, наконец, фаза «На завершении» все еще показывает flowVars.aCounter со значением 1,это было то, что переменная была инициализирована на этапе ввода.

3.0 Случай для записи переменных

Рисунок 3.0a.

Переменная записи определенно недоступна на этапе завершения. Пользователи не могут объявить переменную записи в фазе ввода, причина в том, что записи существуют только после неявной фазы загрузки и отправки. Рисунок 3.0a суммирует доступность переменных записи, но они немного размыты относительно фактической области действия / времени жизни переменной записи.

Чтобы проиллюстрировать вам время жизни переменной записи, я создал следующий поток. Вы можете начать установку переменной записи только в шаге пакета, я сделал это в Batch_step_1 (согласно рисунку 3.0b).

Рисунок 3.0b.

Обратите внимание, что я использовал процессор сообщений выражений вместо процессора сообщений Groovy для установки recordVars.rACounter, причина в том, что оболочка Groovy в Mule не распознает recordVars (обратите внимание, что вам не нужно тратить часы пытаясь понять почему).

Вы не можете получить доступ к переменной записи на этапе завершения, иначе это даст вам исключение времени выполнения. Ниже приведен полный пакетный XML-документ, показывающий, как правильно использовать переменные записи.

    <batch:job name="batchjob_withnoInput">
        <batch:input>
            <logger message="Batch Job Input: #[payload]" level="INFO" doc:name="Logger"/>
            <set-variable variableName="aCounter" value="#[1]" doc:name="flowVars.aCounter = 1"/>
        </batch:input>
        <batch:process-records>
            <batch:step name="Batch_Step_1">
                <batch:set-record-variable variableName="rACounter" value="#[1]" doc:name="recordVars.rACounter = 1"/>
                <scripting:component doc:name="flowVars.aCounter add 1">
                    <scripting:script engine="Groovy"><![CDATA[flowVars.aCounter = flowVars.aCounter + 1]]></scripting:script>
                </scripting:component>
                <expression-component doc:name="recordVars.rACounter add 1"><![CDATA[recordVars.rACounter = recordVars.rACounter + 1]]></expression-component>
                <logger message="Batch Step 1 flowVars.aCounter: #[flowVars.aCounter + &quot;\n&quot;] Batch Step 1 recordVars.rACounter: #[recordVars.rACounter + &quot;\n&quot;]" level="INFO" doc:name="Logger"/>
            </batch:step>
            <batch:step name="Batch_Step_2" >
                <scripting:component doc:name="flowVars.aCounter add 1">
                    <scripting:script engine="Groovy"><![CDATA[flowVars.aCounter = flowVars.aCounter + 1]]></scripting:script>
                </scripting:component>
                <expression-component doc:name="recordVars.rACounter add 1"><![CDATA[recordVars.rACounter = recordVars.rACounter + 1]]></expression-component>
                <logger message="Batch Step 2 flowVars.aCounter: #[flowVars.aCounter + &quot;\n&quot;] Batch Step 2 recordVars.rACounter: #[recordVars.rACounter + &quot;\n&quot;]" level="INFO" doc:name="Logger"/>
            </batch:step>
        </batch:process-records>
        <batch:on-complete>
            <logger message="oncomplete flowVars.aCounter : #[flowVars.aCounter + &quot;\n&quot;]" level="INFO" doc:name="Logger"/>
        </batch:on-complete>
    </batch:job>

Листинг 3.0a.

Фрагмент кода в листинге 3.0a сгенерирует следующий вывод:

[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Created instance a3e26950-9758-11e7-956a-b62d20524153 for batch job batchjob_withnoInput
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Starting input phase
[[batchplayground].HTTP_Listener_Configuration.worker.01] org.mule.api.processor.LoggerMessageProcessor: Batch Job Input: [{OriginNum=1, ID=001}]
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Input phase completed
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.queue.BatchQueueLoader: Starting loading phase for instance 'a3e26950-9758-11e7-956a-b62d20524153' of job 'batchjob_withnoInput'
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.queue.BatchQueueLoader: Finished loading phase for instance a3e26950-9758-11e7-956a-b62d20524153 of job batchjob_withnoInput. 1 records were loaded
[[batchplayground].HTTP_Listener_Configuration.worker.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Started execution of instance 'a3e26950-9758-11e7-956a-b62d20524153' of job 'batchjob_withnoInput'
[batch-job-batchjob_withnoInput-work-manager.02] org.mule.api.processor.LoggerMessageProcessor: Batch Step 1 flowVars.aCounter: 2
                                                                                                Batch Step 1 recordVars.rACounter: 2
[batch-job-batchjob_withnoInput-work-manager.02] com.mulesoft.module.batch.DefaultBatchStep: Step Batch_Step_1 finished processing all records for instance a3e26950-9758-11e7-956a-b62d20524153 of job batchjob_withnoInput
[batch-job-batchjob_withnoInput-work-manager.01] org.mule.api.processor.LoggerMessageProcessor: Batch Step 2 flowVars.aCounter: 2
                                                                                                Batch Step 2 recordVars.rACounter: 3
[batch-job-batchjob_withnoInput-work-manager.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Starting execution of onComplete phase for instance a3e26950-9758-11e7-956a-b62d20524153 of job batchjob_withnoInput
[batch-job-batchjob_withnoInput-work-manager.01] org.mule.api.processor.LoggerMessageProcessor: oncomplete flowVars.aCounter : 1
[batch-job-batchjob_withnoInput-work-manager.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Finished execution of onComplete phase for instance a3e26950-9758-11e7-956a-b62d20524153 of job batchjob_withnoInput
        _withnoInput-work-manager.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: Finished execution for instance 'a3e26950-9758-11e7-956a-b62d20524153' of job 'batchjob_withnoInput'. Total Records processed: 1. Successful records: 1. Failed Records: 0
[batch-job-batchjob_withnoInput-work-manager.01] com.mulesoft.module.batch.engine.DefaultBatchEngine: 
[batch-job-batchjob_withnoInput-work-manager.01] com.mulesoft.module.batch.DefaultBatchStep: Step Batch_Step_2 finished processing all records for instance a3e26950-9758-11e7-956a-b62d20524153 of job batchjob_withnoInput

4.0 Как насчет переменных сеанса в пакетных заданиях?

Если вы наблюдаете как значения «flowVars.aCounter», так и «recordVars.rACounter», вы увидите, что только переменная записи может передавать свое состояние через этапы пакета, но как только она выходит за пределы области пакетного процесса (где все шаг пакета определен), он больше не доступен контексту выполнения.

Рисунок 4.0a.

На рисунке 4.0a показано, что я включил переменную сеанса с полностью определенным именем «sessionVars.SaCounter». Его поведение совпадает с переменной потока. В пакетном задании понятие контекста области видимости для переменных потока и переменных сеанса не работает так, как если бы они находились в потоке.

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

Рисунок 4.0b.

Из журналов вы узнаете, что по сути работают три потока. Установка переменной потока и сеанса происходит в потоке с именем «[batchplayground] .HTTP_Listener_Configuration.worker.01.» Это поток-получатель из потока «batchplaygroundFlow», поскольку этот поток имеет входящую конечную точку HTTP, он автоматически будет потоком со стратегией синхронной обработки. Два других потока, «batch-job-batchjob_withnoInput-work-manager.01» и «batch-job-batchjob_withnoInput-work-manager.02», являются несопоставимыми потоками, выделенными средой выполнения Mule для запуска как пакетного шага 1, так и пакетного шага 2. В этой многопоточной среде переменные потока и переменные сеанса не синхронизируются; синхронизируются только переменные записи.

5.0 Заключение

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

33
Share: