paulwong

如何优雅地停止SPRING BATCH中的REMOTE CHUNKING JOB

SPRING BATCH中的REMOTE CHUNKING JOB,由于是基于MASTER/SLAVE的架构,其中某个STEP是会在远程机器中执行,如果要停止这个JOB,需要考虑两个问题:
1、什么时候发出停止指令
2、如何等待远程STEP的完成

一般停止JOB,可用JobOperator.stop(long executionId)来停止,但这个无法确定什么时候发出停止指令,如果是在CHUNK的处理中途发出,则会出现回滚的现象。
BATCH_STEP_EXECUTION thead tr {background-color: ActiveCaption; color: CaptionText;} th, td {vertical-align: top; font-family: "Tahoma", Arial, Helvetica, sans-serif; font-size: 8pt; padding: 4px; } table, td {border: 1px solid silver;} table {border-collapse: collapse;} thead .col0 {width: 173px;} .col0 {text-align: right;} thead .col1 {width: 82px;} .col1 {text-align: right;} thead .col2 {width: 282px;} thead .col3 {width: 164px;} .col3 {text-align: right;} thead .col4 {width: 161px;} thead .col5 {width: 161px;} thead .col6 {width: 109px;} thead .col7 {width: 127px;} .col7 {text-align: right;} thead .col8 {width: 109px;} .col8 {text-align: right;} thead .col9 {width: 118px;} .col9 {text-align: right;} thead .col10 {width: 117px;} .col10 {text-align: right;} thead .col11 {width: 142px;} .col11 {text-align: right;} thead .col12 {width: 150px;} .col12 {text-align: right;} thead .col13 {width: 166px;} .col13 {text-align: right;} thead .col14 {width: 137px;} .col14 {text-align: right;} thead .col15 {width: 109px;} thead .col16 {width: 156px;} thead .col17 {width: 161px;}
STEP_EXECUTION_ID VERSION STEP_NAME JOB_EXECUTION_ID START_TIME END_TIME STATUS COMMIT_COUNT READ_COUNT FILTER_COUNT WRITE_COUNT READ_SKIP_COUNT WRITE_SKIP_COUNT PROCESS_SKIP_COUNT ROLLBACK_COUNT EXIT_CODE EXIT_MESSAGE LAST_UPDATED
2304 169 step2HandleXXX 434 2020-06-22 16:27:54 2020-06-22 16:32:46 STOPPED 167 5010 0 4831 0 155 0 161 STOPPED org.springframework.batch.core.JobInterruptedException 2020-06-22 16:32:46


另外SPRING BATCH也不会等远程STEP执行完成,就将JOB的状态设为Complete。

发出停止的指令应通过ChunkListener达成:

public class ItemMasterChunkListener extends ChunkListenerSupport{
    
    private static final Logger log = LoggerFactory.getLogger(ItemMasterChunkListener.class);
    
    
    @Override
    public void beforeChunk(ChunkContext context) {
        log.info("ItemMasterProcessor.beforeChunk");
    }


    @Override
    public void afterChunk(ChunkContext context) {
        log.info("ItemMasterProcessor.afterChunk");
        if(XXXX.isStoppingOrPausing()) {
            log.info("context.getStepContext().getStepExecution().setTerminateOnly()");
            context.getStepContext().getStepExecution().setTerminateOnly();
        }
    }


    @Override
    public void afterChunkError(ChunkContext context) {
        log.info("ItemMasterProcessor.afterChunkError");
    }


}


配置BEAN:

@Bean
@StepScope
public ItemMasterChunkListener novaXItemMasterChunkListener() {
     return new ItemMasterChunkListener();
}
    
this.masterStepBuilderFactory
                    .<X, X>get("step2Handle")
                    .listener(itemMasterChunkListener())
                    .build();


由于是在CHUNK完成的时候发出停止指令,就不会出现ROLLBACK的情况。

等待远程STEP完成,通过读取MQ上的MESSAGE是否被消费完成,PENDDING的MESSAGE为0的条件即可。

public class JobExecutionListenerSupport implements JobExecutionListener {

    /* (non-Javadoc)
     * @see org.springframework.batch.core.domain.JobListener#afterJob()
     
*/
    @Override
    public void afterJob(JobExecution jobExecution) {
        Integer totalPendingMessages = 0;
        String queueName = "";
        
        
        String messageSelector = "JOB_EXECUTION_ID=" + jobExecution.getJobInstance().getInstanceId();
        do{
            totalPendingMessages = 
                    this.jmsTemplate.browseSelected(queueName, messageSelector, 
                                (session, browser) -> 
                                    Collections.list(browser.getEnumeration()).size()
                            );
            
            String brokerURL = null;
            if(jmsTemplate.getConnectionFactory() instanceof JmsPoolConnectionFactory) {
                JmsPoolConnectionFactory connectionFactory =
                        (JmsPoolConnectionFactory)jmsTemplate.getConnectionFactory();
                ActiveMQConnectionFactory activeMQConnectionFactory =
                        (ActiveMQConnectionFactory)connectionFactory.getConnectionFactory();
                brokerURL = activeMQConnectionFactory.getBrokerURL();
            } else if(jmsTemplate.getConnectionFactory() instanceof CachingConnectionFactory) {
                CachingConnectionFactory connectionFactory =
                        (CachingConnectionFactory)jmsTemplate.getConnectionFactory();
                ActiveMQConnectionFactory activeMQConnectionFactory =
                        (ActiveMQConnectionFactory)connectionFactory.getTargetConnectionFactory();
                brokerURL = activeMQConnectionFactory.getBrokerURL();
            }
            
            LOGGER.info("queueName = {}, {}, totalPendingMessages = {}, url={}", 
                    queueName, messageSelector, totalPendingMessages, brokerURL);
            Assert.notNull(totalPendingMessages, "totalPendingMessages must not be null.");
            try {
                Thread.sleep(5_000);
            } catch (InterruptedException e) {
                LOGGER.error(e.getMessage(), e);
            }
        } while(totalPendingMessages.intValue() > 0);
        
    }

    /* (non-Javadoc)
     * @see org.springframework.batch.core.domain.JobListener#beforeJob(org.springframework.batch.core.domain.JobExecution)
     
*/
    @Override
    public void beforeJob(JobExecution jobExecution) {
    }

}


这样整个JOB就能无异常地停止,且会等待远程STEP完成。

Reference:
https://docs.spring.io/spring-batch/docs/4.1.3.RELEASE/reference/html/common-patterns.html#stoppingAJobManuallyForBusinessReasons

https://stackoverflow.com/questions/13603949/count-number-of-messages-in-a-jms-queue

https://stackoverflow.com/questions/55499965/spring-batch-stop-job-execution-from-external-class

https://stackoverflow.com/questions/34621885/spring-batch-pollable-channel-with-replies-contains-chunkresponses-even-if-job


posted on 2020-06-23 11:00 paulwong 阅读(770) 评论(0)  编辑  收藏 所属分类: SPRING BOOT


只有注册用户登录后才能发表评论。


网站导航: