专栏名称: ImportNew
伯乐在线旗下账号,专注Java技术分享,包括Java基础技术、进阶技能、架构设计和Java技术领域动态等。
目录
相关文章推荐
芋道源码  ·  孤陋寡闻了,原来 MySQL 还能这么写? ·  昨天  
芋道源码  ·  魔改 Arthas,又一款开源的线上问题定位神器 ·  3 天前  
51好读  ›  专栏  ›  ImportNew

SpringBatch 中的 retry 和 skip 机制实现分析

ImportNew  · 公众号  · Java  · 2017-04-12 12:05

正文

(点击上方公众号,可快速关注)


来源:fangjian0423,

fangjian0423.github.io/2016/11/09/springbatch-retry-skip/#more

如有好文章投稿,请点击 → 这里了解详情


SpringBatch是spring框架下的一个子模块,用于处理批处理的批次框架。本文主要分析SpringBatch中的retry和skip机制的实现。


先简单说明下SpringBatch在SpringBoot中的使用。


如果要在springboot中使用batch的话,直接加入以下依赖即可:


    org.springframework.boot

    spring-boot-starter-batch


然后使用注解开启Batch模块:


...

@EnableBatchProcessing

public class Application { ... }


之后就可以注入JobBuilderFactory和StepBuilderFactory:


@Autowired

private JobBuilderFactory jobs;

 

@Autowired

private StepBuilderFactory steps;


有了这2个factory之后,就可以build job。


SpringBatch中的相关基础概念比如ItemReader、ItemWriter、Chunk等本文就不介绍了。


我们以FlatFileItemReader作为reader,一个自定义Writer用于打印reader中读取出来的数据。


这个定义的writer遇到good job这条数据的时候会报错,具体逻辑如下:


@Override

public void write(List extends String> items) throws Exception {

    System.out.println("handle start =====" + items);

    for(String a : items) {

        if(a.equals("good job")) {

            throw new Exception("custom exception");

        }

    }

    System.out.println("handle end.. -----" + items);

}


其中reader中读取的文件中的数据如下:


hello world

hello coder

good job

cool

66666


我们使用StepBuilderFactory构造Step,chunkSize设置为2。然后在job1中使用并执行:


stepBuilderFactory.get("test-step").chunk(2).reader(reader).writer(writer).build();


执行job1后console打印如下:


handle start =====[hello world, hello coder]

handle end.. -----[hello world, hello coder]

handle start =====[good job, cool]


job1遇到了good job这条数据,writer抛出了异常,由于没有使用skip或者retry机制,导致整个流程停止。job1的处理流程底层在SimpleChunkProcessor这个类中完成,包括processor、writer的使用。


接下里我们构造一个job2,job2使用skip机制(其中skipLimit必须要和skip(Class extends Throwable> type)一起使用),skip机制可以防止writer发生异常后不停止整个job,但是需要同时满足skip的限制次数和skip对应的Exception是发生异常的父类或自身关系条件才不会停止整个job,这里我们使用Exception作为异常和Integer.MAX_VALUE作为skip的限制次数为例:


stepBuilderFactory.get.get("test-step").chunk(2).reader(reader).writer(writer).faultTolerant().skipLimit(Integer.MAX_VALUE).skip(Exception.class).build();


执行job2 后console打印如下:


handle start =====[hello world, hello coder]

handle end.. -----[hello world, hello coder]

handle start =====[good job, cool]

handle start =====[good job]

handle start =====[cool]

handle end.. -----[cool]

handle start =====[66666]

handle end.. -----[66666]


我们看到good job这条数据发生的异常被skip掉了,job完整的执行。


但是发现了另外一个问题,那就是处理 [good job, cool] 这批数据的时候,发生了异常,但是接下来执行了 [good job] 和 [cool] 这两批chunk为1的批次。这是在ItemWriter中执行的,它也会在ItemWriteListener中执行多次。


换句话说,如果使用了skip功能,那么对于需要被skip的批次数据中会进行scan操作找出具体是哪1条数据的原因,这里的scan操作指的是一条一条数据的遍历。


这个过程为什么叫scan呢? 在源码中,FaultTolerantChunkProcessor类(处理带有skip或者retry机制的处理器,跟SimpleChunkProcessor类似,只不过SimpleChunkProcessor处理简单的Job)里有个私有方法scan:


private void scan(final StepContribution contribution, final Chunk inputs, final Chunk outputs,

        ChunkMonitor chunkMonitor, boolean recovery) throws Exception {

 

    ...

 

    Chunk.ChunkIterator inputIterator = inputs.iterator();

    Chunk.ChunkIterator outputIterator = outputs.iterator();

 

    List items = Collections.singletonList(outputIterator.next()); // 拿出需要写的数据中的每一条数据

    inputIterator.next();

    try {

        writeItems(items); // 写每条数据

        doAfterWrite(items);

        contribution.incrementWriteCount(1);

        inputIterator.remove();

        outputIterator.remove();

    }

    catch (Exception e) { // 写的时候如果发生了异常

        doOnWriteError(e, items);

        if (!shouldSkip(itemWriteSkipPolicy, e, -1) && !rollbackClassifier.classify(e)) {

            inputIterator.remove();

            outputIterator.remove();

        }

        else {

            // 具体的skip策略

            checkSkipPolicy(inputIterator, outputIterator, e, contribution, recovery);

        }

        if (rollbackClassifier.classify(e)) {

            throw e;

        }

    }

    chunkMonitor.incrementOffset();

    if (outputs.isEmpty()) { // 批次里的所有数据处理完毕之后 scanning 设置为false

        data.scanning(false);

        inputs.setBusy(false);

        chunkMonitor.resetOffset();

    }

}


这个scan方法触发的条件是UserData这个内部类里的scanning被设置为true,这里被设置为true是在处理批次数据出现异常后并且不能retry的情况下才会被设置的。


try {

    batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(inputs,

            rollbackClassifier));

}

catch (Exception e) {

    RetryContext context = contextHolder.get();

     if (!batchRetryTemplate.canRetry(context)) {

         // 设置scanning为true

        data.scanning(true);

    }

    throw e;

}


这就是为什么skip机制在skip数据的时候会去scan批次中的每条数据,然后并找出需要被skip的数据的原理。


job3带有retry功能,retry的功能在于出现某个异常并且这个异常可以被retry所接受的话会进行retry,retry的次数可以进行配置,我们配置了3次retry:


stepBuilderFactory.get.get("test-step").chunk(2).reader(reader).writer(writer).faultTolerant().skipLimit(Integer.MAX_VALUE).skip(Exception.class).retryLimit(3).retry(Exception.class).build();


执行 job3后console打印如下:


handle start =====[hello world, hello coder]

handle end.. -----[hello world, hello coder]

handle start =====[good job, cool]

handle start =====[good job, cool]

handle start =====[good job, cool]

handle start =====[good job]

handle start =====[cool]

handle end.. -----[cool]

handle start =====[66666]

handle end.. -----[66666]


[good job, cool] 这批数据retry了3次,而且都失败了。失败之后进行了skip操作。


SpringBatch中的retry和skip都有对应的policy实现,默认的retry policy是SimpleRetryPolicy,可以设置retry次数和接收的exception。比如可以使用NeverRetryPolicy:


.retryPolicy(new NeverRetryPolicy())


使用NeverRetryPolicy之后,便不再retry了,只会skip。SpringBatch内部的retry是使用Spring的retry模块完成的。执行的时候可以设置RetryCallback和RecoveryCallback。


SpringBatch中默认的skip policy是LimitCheckingItemSkipPolicy。


参考资料:


  • http://stackoverflow.com/questions/16567432/how-is-the-skipping-implemented-in-spring-batch


  • http://docs.spring.io/spring-batch/reference/html/retry.html


  • https://github.com/spring-projects/spring-retry


看完本文有收获?请转发分享给更多人

关注「ImportNew」,看技术干货