其中有几个细节，此处不再赘述。

1. 如何控制一次读取几个item？
2. 如何控制最后一行读完就不读了？
3. 如果需要跳过文件头的前N行，怎么处理？
4. 在StepContribution中记录读取数量

org.springframework.batch.core.step.item.SimpleChunkProcessor#process   @Nullable  @Override  public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {    @SuppressWarnings("unchecked")   Chunk<I> inputs = (Chunk<I>) chunkContext.getAttribute(INPUTS_KEY);   if (inputs == null) {    inputs = chunkProvider.provide(contribution);    if (buffering) {     chunkContext.setAttribute(INPUTS_KEY, inputs);    }   }    chunkProcessor.process(contribution, inputs);   chunkProvider.postProcess(contribution, inputs);    // Allow a message coming back from the processor to say that we   // are not done yet   if (inputs.isBusy()) {    logger.debug("Inputs still busy");    return RepeatStatus.CONTINUABLE;   }    chunkContext.removeAttribute(INPUTS_KEY);   chunkContext.setComplete();    if (logger.isDebugEnabled()) {    logger.debug("Inputs not busy, ended: " + inputs.isEnd());   }   return RepeatStatus.continueIf(!inputs.isEnd());   } 

3.3何时调用Processor/Writer？

在RepeatTemplate和外围事务模板的包装下，通过SimpleChunkProcessor进行处理：

1. 查出若干条数的items，打包为Chunk
2. 遍历items，逐个item调用processor
   1. 通知StepListener，环绕处理调用before/after方法

    // 忽略无关代码...  @Override  public final void process(StepContribution contribution, Chunk<I> inputs) throws Exception {    // 输入为空，直接返回If there is no input we don't have to do anything more   if (isComplete(inputs)) {    return;   }    // Make the transformation, calling remove() on the inputs iterator if   // any items are filtered. Might throw exception and cause rollback.   Chunk<O> outputs = transform(contribution, inputs);    // Adjust the filter count based on available data   contribution.incrementFilterCount(getFilterCount(inputs, outputs));    // Adjust the outputs if necessary for housekeeping purposes, and then   // write them out...   write(contribution, inputs, getAdjustedOutputs(inputs, outputs));   }      // 遍历items，逐个item调用processor  protected Chunk<O> transform(StepContribution contribution, Chunk<I> inputs) throws Exception {   Chunk<O> outputs = new Chunk<>();   for (Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {    final I item = iterator.next();    O output;    String status = BatchMetrics.STATUS_SUCCESS;    try {     output = doProcess(item);    }    catch (Exception e) {     /*      * For a simple chunk processor (no fault tolerance) we are done here, so prevent any more processing of these inputs.      */     inputs.clear();     status = BatchMetrics.STATUS_FAILURE;     throw e;    }    if (output != null) {     outputs.add(output);    }    else {     iterator.remove();    }   }   return outputs;  }  

4. 每个step是如何与事务处理挂钩？

在TaskletStep#doExecute中会使用TransactionTemplate，包装事务操作

标准的事务操作，通过函数式编程风格，从action的CallBack调用实际处理方法

1. 通过transactionManager获取事务
2. 执行操作
3. 无异常，则提交事务
4. 若异常，则回滚

    // org.springframework.batch.core.step.tasklet.TaskletStep#doExecute     result = new TransactionTemplate(transactionManager, transactionAttribute)         .execute(new ChunkTransactionCallback(chunkContext, semaphore));      // 事务启用过程     // org.springframework.transaction.support.TransactionTemplate#execute  @Override  @Nullable  public <T> T execute(TransactionCallback<T> action) throws TransactionException {   Assert.state(this.transactionManager != null, "No PlatformTransactionManager set");    if (this.transactionManager instanceof CallbackPreferringPlatformTransactionManager) {    return ((CallbackPreferringPlatformTransactionManager) this.transactionManager).execute(this, action);   }   else {    TransactionStatus status = this.transactionManager.getTransaction(this);    T result;    try {     result = action.doInTransaction(status);    }    catch (RuntimeException | Error ex) {     // Transactional code threw application exception -> rollback     rollbackOnException(status, ex);     throw ex;    }    catch (Throwable ex) {     // Transactional code threw unexpected exception -> rollback     rollbackOnException(status, ex);     throw new UndeclaredThrowableException(ex, "TransactionCallback threw undeclared checked exception");    }    this.transactionManager.commit(status);    return result;   }  } 

5. 怎么控制每个chunk几条记录提交一次事务？ 控制每个事务窗口处理的item数量

在配置任务时，有个step级别的参数，[commit-interval]，用于每个事务窗口提交的控制被处理的item数量。

RepeatTemplate#executeInternal 在处理单条item后，会查看已处理完的item数量，与配置的chunk数量做比较，如果满足chunk数，则不再继续，准备提交事务。

StepBean在初始化时，会新建SimpleCompletionPolicy(chunkSize会优先使用配置值，默认是5)

在每个chunk处理开始时，都会调用SimpleCompletionPolicy#start新建RepeatContextSupport#count用于计数。

源码（简化） org.springframework.batch.repeat.support.RepeatTemplate#executeInternal

 /**  * Internal convenience method to loop over interceptors and batch  * callbacks.  * @param callback the callback to process each element of the loop.  */ private RepeatStatus executeInternal(final RepeatCallback callback) {  // Reset the termination policy if there is one...        // 此处会调用completionPolicy.start方法，更新chunk的计数器  RepeatContext context = start();  // Make sure if we are already marked complete before we start then no processing takes place.        // 通过running字段来判断是否继续处理next  boolean running = !isMarkedComplete(context);        // 省略listeners处理....  // Return value, default is to allow continued processing.  RepeatStatus result = RepeatStatus.CONTINUABLE;  RepeatInternalState state = createInternalState(context);  try {   while (running) {    /*     * Run the before interceptors here, not in the task executor so     * that they all happen in the same thread - it's easier for     * tracking batch status, amongst other things.     */                // 省略listeners处理....    if (running) {     try {                        // callback是实际处理方法，类似函数式编程      result = getNextResult(context, callback, state);      executeAfterInterceptors(context, result);     }     catch (Throwable throwable) {      doHandle(throwable, context, deferred);     }                    // 检查当前chunk是否处理完，决策出是否继续处理下一条item     // N.B. the order may be important here:     if (isComplete(context, result) || isMarkedComplete(context) || !deferred.isEmpty() {      running = false;     }    }   }   result = result.and(waitForResults(state));            // 省略throwables处理....   // Explicitly drop any references to internal state...   state = null;  }  finally {            // 省略代码...  }  return result; } 

总结

JSR-352标准定义了Java批处理的基本模型，包含批处理的元数据像 JobExecutions，JobInstances，StepExecutions 等等。通过此类模型，提供了许多基础组件与扩展点：

1. 完善的基础组件
   1. Spring Batch 有很多的这类组件 例如 ItemReaders，ItemWriters，PartitionHandlers 等等对应各类数据和环境。
2. 丰富的配置
   1. JSR－352 定义了基于XML的任务设置模型。Spring Batch 提供了基于Java （类型安全的）的配置方式
3. 可伸缩性
   1. 伸缩性选项－Local Partitioning 已经包含在JSR －352 里面了。但是还应该有更多的选择 ，例如Spring Batch 提供的 Multi－threaded Step，Remote Partitioning ，Parallel Step，Remote Chunking 等等选项
4. 扩展点
   1. 良好的listener模式，提供step/job运行前后的锚点，以供开发人员个性化处理批处理流程。

2013年， JSR-352标准包含在 JavaEE7中发布，到2022年已近10年，Spring也在探索新的批处理模式， 如Spring Attic /Spring Cloud Data Flow。 https://docs.spring.io/spring-batch/docs/current/reference/html/jsr-352.html

扩展

1. Job/Step运行时的上下文，是如何保存？如何控制？

整个Job在运行时，会将运行信息保存在JobContext中。 类似的，Step运行时也有StepContext。可以在Context中保存一些参数，在任务或者步骤中传递使用。

查看JobContext/StepContext源码，发现仅用了普通变量保存Execution，这个类肯定有线程安全问题。 生产环境中常常出现多个任务并处处理的情况。

SpringBatch用了几种方式来包装并发安全：

1. 每个job初始化时，通过JobExecution新建了JobContext，每个任务线程都用自己的对象。
2. 使用JobSynchronizationManager，内含一个ConcurrentHashMap，KEY是JobExecution，VALUE是JobContext
3. 在任务解释时，会移除当前JobExecution对应的k-v

此处能看到，如果在JobExecution存储大量的业务数据，会导致无法GC回收，导致OOM。所以在上下文中，只应保存精简的数据。

2. step执行时，如果出现异常，如何保护运行状态？

在源码中，使用了各类同步控制和加锁、oldVersion版本拷贝，整体比较复杂(org.springframework.batch.core.step.tasklet.TaskletStep.ChunkTransactionCallback#doInTransaction)

1. oldVersion版本拷贝：上一次运行出现异常时，本次执行时沿用上次的断点内容

// 节选部分代码 oldVersion = new StepExecution(stepExecution.getStepName(), stepExecution.getJobExecution()); copy(stepExecution, oldVersion);  private void copy(final StepExecution source, final StepExecution target) {  target.setVersion(source.getVersion());  target.setWriteCount(source.getWriteCount());  target.setFilterCount(source.getFilterCount());  target.setCommitCount(source.getCommitCount());  target.setExecutionContext(new ExecutionContext(source.getExecutionContext())); } 

2. 信号量控制，在每个chunk运行完成后，需先获取锁，再更新stepExecution前
   1. Shared semaphore per step execution, so other step executions can run in parallel without needing the lockSemaphore (org.springframework.batch.core.step.tasklet.TaskletStep#doExecute)

// 省略无关代码 try {  try {         // 执行w-p-r模型方法   result = tasklet.execute(contribution, chunkContext);   if (result == null) {    result = RepeatStatus.FINISHED;   }  }  catch (Exception e) {   // 省略...  } } finally {  // If the step operations are asynchronous then we need to synchronize changes to the step execution (at a  // minimum). Take the lock *before* changing the step execution.  try {         // 获取锁   semaphore.acquire();   locked = true;  }  catch (InterruptedException e) {   logger.error("Thread interrupted while locking for repository update");   stepExecution.setStatus(BatchStatus.STOPPED);   stepExecution.setTerminateOnly();   Thread.currentThread().interrupt();  }  stepExecution.apply(contribution); } stepExecutionUpdated = true; stream.update(stepExecution.getExecutionContext()); try {     // 更新上下文、DB中的状态  // Going to attempt a commit. If it fails this flag will stay false and we can use that later.  getJobRepository().updateExecutionContext(stepExecution);  stepExecution.incrementCommitCount();  getJobRepository().update(stepExecution); } catch (Exception e) {  // If we get to here there was a problem saving the step execution and we have to fail.  String msg = "JobRepository failure forcing rollback";  logger.error(msg, e);  throw new FatalStepExecutionException(msg, e); }