[SPRING] 원격 파티션 - 노예 욕심

원격 파티션 - 노예 욕심

다음은 우리가 달성하고자하는 것입니다.

우리는 큰 xml 파일이 다른 VM에서 병렬로 데이터베이스에 준비되기를 원합니다. 이를 위해 확장 가능한 스프링 배치 원격 파티션 접근 방식을 사용하고 있으며 몇 가지 문제가 있습니다. 다음은 높은 수준의 설정입니다.

위의 설정으로

주인은 모든 노예가 완료되기를 기다리지 않고 있으며 알 수없는 상태가 될 것입니다.

org.springframework.batch.core.JobExecutionException: Partition handler returned an unsuccessful step
at org.springframework.batch.core.partition.support.PartitionStep.doExecute(PartitionStep.java:112)
at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:200)
at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)

그리드 크기가 2라면 슬레이브 1은 두 가지를 모두 선택하고, 슬레이브 2는 그 중 하나를 얻지 못합니다 (따라서 욕심 많은 노예). 슬레이브 1은 병렬로 처리하지만 작업은 알 수없는 상태가됩니다.

다음은 우리의 질문입니다.

다음은 우리의 구성입니다.

 <?xml version="1.0" encoding="UTF-8"?>
    <step id="remotePartitionStagingStep">
     <partition partitioner="xmlPartitioner" handler="partitionStagingHandler"/>
       <listeners>
         <listener ref="jobListener" />
       </listeners>
     </step>
     <!-- XML Partitioner starts here -->
     <beans:bean id="xmlPartitioner" class="XMLPartitioner" scope="step">           <beans:property name="partitionThreadName" value="ImportXMLPartition-"/>
      <beans:property name="resource" value="file:///#{jobParameters[ImportFileProcessed]}"/>
            <beans:property name="rootElementNode" value="#{jobParameters[ImportFragmentRootNameWithPrefix]}"/>
        </beans:bean>
       <beans:bean id="partitionStagingHandler" class="org.springframework.batch.integration.partition.MessageChannelPartitionHandler">
            <beans:property name="stepName" value="slave.ImportStagingStep"/>
            <beans:property name="gridSize" value="3"/>
            <beans:property name="replyChannel" ref="aggregatedReplyChannel"/>
            <beans:property name="jobExplorer" ref="jobExplorer"/>
            <beans:property name="messagingOperations">
                <beans:bean class="org.springframework.integration.core.MessagingTemplate">
                    <beans:property name="defaultChannel" ref="requestsChannel"/>
                    <beans:property name="receiveTimeout" value="${batch.gateway.receiveTimeout}" />  //360000 is the current value
                </beans:bean>
            </beans:property>
        </beans:bean>
         <int:aggregator ref="partitionStagingHandler"
                        input-channel="replyChannel"
                        output-channel="aggregatedReplyChannel"
                        send-timeout="${batch.gateway.receiveTimeout}"
                        expire-groups-upon-timeout="true"/>
         <int:channel id="requestsChannel">
            <int:interceptors>
                <int:wire-tap channel="logChannel"/>
            </int:interceptors>
        </int:channel>
         <int-jms:outbound-channel-adapter connection-factory="connectionFactory"
                                          channel="requestsChannel"
                                          destination-name="requestsQueue"/>
        <int:channel id="aggregatedReplyChannel">
            <int:queue/>
        </int:channel>
         <int:channel id="replyChannel">
            <int:interceptors>
                <int:wire-tap channel="logChannel"/>
            </int:interceptors>
        </int:channel>
         <int-jms:message-driven-channel-adapter connection-factory="connectionFactory"
                                                channel="replyChannel"
                                                error-channel="errorChannel"
                                                destination-name="replyQueue"/>

    <step id="slave.ImportStagingStep">
    <tasklet transaction-manager="transactionManager">
    <chunk reader="StagingSpecificItemReader" processor="chainedStagingProcessor" writer="StagingItemWriter"
             commit-interval="${import.CommitInterval}" skip-limit="${import.skipLimit}" retry-policy="neverRetryPolicy">
     <streams>
      <stream ref="errorFlatFileRecordWriter"/>
     </streams>
     <skippable-exception-classes>
       <include class="java.lang.Exception"/>
       <exclude class="org.springframework.oxm.UnmarshallingFailureException"/>
       <exclude class="java.lang.Error"/>
     </skippable-exception-classes>
     </chunk>
     <listeners>
        <listener ref="stepExceptionListener"/>
        <listener ref="stagingListener"/>
     </listeners>
       </tasklet>
       <listeners>
          <listener ref="slaveStepExecutionListener"/>
       </listeners>
      </step>
        <beans:bean id="slaveStepExecutionListener" class="StepExecutionListener"></beans:bean>
        <!-- JMS config for staging step starts here -->
        <int:channel id="replyChannel">
            <int:interceptors>
                <int:wire-tap channel="logChannel"/>
            </int:interceptors>
        </int:channel>
            <int:channel id="requestChannel">
            <int:interceptors>
                <int:wire-tap channel="logChannel"/>
            </int:interceptors>
        </int:channel>
        <int-jms:message-driven-channel-adapter connection-factory="connectionFactory"
                                                destination-name="requestsQueue"
                                                error-channel="errorChannel"
                                                channel="requestsChannel"/>

        <int-jms:outbound-channel-adapter connection-factory="connectionFactory"
                                          destination-name="replyQueue"
                                          channel="replyChannel"/>

        <int:service-activator input-channel="requestsChannel"
                               output-channel="replyChannel"
                               ref="stepExecutionRequestHandler"/>

        <!-- JMS config for staging step ends here -->

        <!-- The logChannel is configured as an interceptor to channels so that messages are logged. -->
        <int:logging-channel-adapter auto-startup="true" log-full-message="true" id="logChannel" level="INFO"/>

        <int:channel id="errorChannel" />

        <int:service-activator input-channel="errorChannel" method="handleException">
          <beans:bean class="ErrorHandler" />
        </int:service-activator>

해결법

1. ==============================

   1.우리가 가진 문제를 발견했습니다.

   우리가 가진 문제를 발견했습니다.

   슬레이브는 스프링 배치 작업 정의를 두 번 포함하여 각 슬레이브에 대해 두 번 등록되는 소비자 수를 발생 시켰습니다. 나는 하나의 파티션이 하나의 슬레이브로가는 한,이 시점에서 슬레이브에서 동시에 두 개의 파티션이 동시에 실행될 때 슬레이브에서 마스터로 메시지를 다시 라우팅하는 일부 구성이 누락되었다고 생각합니다. 노예) 우리는 괜찮아요.

from https://stackoverflow.com/questions/50826773/remote-partition-slave-getting-greedy by cc-by-sa and MIT license