源码解析springbatch的job是如何运行的?

202208-源码解析springbatch的job是如何运行的?

注,本文中的demo代码节选于图书《Spring Batch批处理框架》的配套源代码,并做并适配springboot升级版本,完全开源。

SpringBatch的背景和用法,就不再赘述了,默认本文受众都使用过batch框架。
本文仅讨论普通的ChunkStep,分片/异步处理等功能暂不讨论。

1. 表结构

Spring系列的框架代码,大多又臭又长,让人头晕。先列出整体流程,再去看源码。顺带也可以了解存储表结构。

  1. 每一个jobname,加运行参数的MD5值,被定义为一个job_instance,存储在batch_job_instance表中;
  2. job_instance每次运行时,会创建一个新的job_execution,存储在batch_job_execution / batch_job_execution_context 表中;
    1. 扩展:任务重启时,如何续作? 答,判定为任务续作,创建新的job_execution时,会使用旧job_execution的运行态ExecutionContext(通俗讲,火车出故障只换了车头,车厢货物不变。)
  3. job_execution会根据job排程中的step顺序,逐个执行,逐个转化为step_execution,并存储在batch_step_execution / batch_step_execution_context表中
  4. 每个step在执行时,会维护step运行状态,当出现异常或者整个step清单执行完成,会更新job_execution的状态
  5. 在每个step执行前后、job_execution前后,都会通知Listener做回调。

框架使用的表

batch_job_instance
batch_job_execution
batch_job_execution_context
batch_job_execution_params
batch_step_execution
batch_step_execution_context
batch_job_seq
batch_step_execution_seq
batch_job_execution_seq

2. API入口

先看看怎么调用启动Job的API,看起来非常简单,传入job信息和参数即可

    @Autowired
    @Qualifier("billJob")
    private Job job;
    
    @Test
    public void billJob() throws Exception {
        JobParameters jobParameters = new JobParametersBuilder()
                .addLong("currentTimeMillis", System.currentTimeMillis())
                .addString("batchNo","2022080402")
                .toJobParameters();
        JobExecution result = jobLauncher.run(job, jobParameters);
        System.out.println(result.toString());

        Thread.sleep(6000);
    }
    
    
        
            
                
                
            
        
    

org.springframework.batch.core.launch.support.SimpleJobLauncher#run

// 简化部分代码(参数检查、log日志)
@Override
public JobExecution run(final Job job, final JobParameters jobParameters){
	final JobExecution jobExecution;
	JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);
       // 上次执行存在,说明本次请求是重启job,先做检查
	if (lastExecution != null) {
		if (!job.isRestartable()) {
			throw new JobRestartException("JobInstance already exists and is not restartable");
		}
		/* 检查stepExecutions的状态
		 * validate here if it has stepExecutions that are UNKNOWN, STARTING, STARTED and STOPPING
		 * retrieve the previous execution and check
		 */
		for (StepExecution execution : lastExecution.getStepExecutions()) {
			BatchStatus status = execution.getStatus();
			if (status.isRunning() || status == BatchStatus.STOPPING) {
				throw new JobExecutionAlreadyRunningException("A job execution for this job is already running: "
						+ lastExecution);
			} else if (status == BatchStatus.UNKNOWN) {
				throw new JobRestartException(
						"Cannot restart step [" + execution.getStepName() + "] from UNKNOWN status. ");
			}
		}
	}
	// Check jobParameters
	job.getJobParametersValidator().validate(jobParameters);
       // 创建JobExecution 同一个job+参数,只能有一个Execution执行器
	jobExecution = jobRepository.createJobExecution(job.getName(), jobParameters);
	try {
           // SyncTaskExecutor 看似是异步,实际是同步执行(可扩展)
		taskExecutor.execute(new Runnable() {
			@Override
			public void run() {
				try {
                       // 关键入口,请看[org.springframework.batch.core.job.AbstractJob#execute]
					job.execute(jobExecution);
					if (logger.isInfoEnabled()) {
						Duration jobExecutionDuration = BatchMetrics.calculateDuration(jobExecution.getStartTime(), jobExecution.getEndTime());
					}
				}
				catch (Throwable t) {
					rethrow(t);
				}
			}
			private void rethrow(Throwable t) {
                   // 省略各类抛异常
				throw new IllegalStateException(t);
			}
		});
	}
	catch (TaskRejectedException e) {
        // 更新job_execution的运行状态
		jobExecution.upgradeStatus(BatchStatus.FAILED);
		if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) {
			jobExecution.setExitStatus(ExitStatus.FAILED.addExitDescription(e));
		}
		jobRepository.update(jobExecution);
	}
	return jobExecution;
}

3. 深入代码流程

简单看看API入口,子类划分较多,继续往后看

总体代码流程

  1. org.springframework.batch.core.launch.support.SimpleJobLauncher#run 入口api,构建jobExecution
  2. org.springframework.batch.core.job.AbstractJob#execute 对jobExecution进行执行、listener的前置处理
  3. FlowJob#doExecute -> SimpleFlow#start 按顺序逐个处理Step、构建stepExecution
  4. JobFlowExecutor#executeStep -> SimpleStepHandler#handleStep -> AbstractStep#execute 执行stepExecution
  5. TaskletStep#doExecute 通过RepeatTemplate,调用TransactionTemplate方法,在事务中执行
    1. 内部类TaskletStep.ChunkTransactionCallback#doInTransaction
  6. 反复调起ChunkOrientedTasklet#execute 去执行read-process-writer方法,
    1. 通过自定义的Reader得到inputs,例如本文实现的是flatReader读取csv文件
    2. 遍历inputs,将item逐个传入,调用processor处理
    3. 调用writer,将outputs一次性写入
    4. 不同reader的实现内容不同,通过缓存读取的行数等信息,可做到分片、按数量处理chunk

JobExecution的处理过程

org.springframework.batch.core.job.AbstractJob#execute


/** 运行给定的job,处理全部listener和DB存储的调用
* Run the specified job, handling all listener and repository calls, and
* delegating the actual processing to {@link #doExecute(JobExecution)}.
*
* @see Job#execute(JobExecution)
* @throws StartLimitExceededException
*             if start limit of one of the steps was exceeded
*/
@Ovrride
public final void execute(JobExecution execution) {

    // 同步控制器,防并发执行
    JobSynchronizationManager.register(execution);
    // 计时器,记录耗时
    LongTaskTimer longTaskTimer = BatchMetrics.createLongTaskTimer("job.active", "Active jobs",
            Tag.of("name", execution.getJobInstance().getJobName()));
    LongTaskTimer.Sample longTaskTimerSample = longTaskTimer.start();
    Timer.Sample timerSample = BatchMetrics.createTimerSample();

    try {
        // 参数再次进行校验
        jobParametersValidator.validate(execution.getJobParameters());

        if (execution.getStatus() != BatchStatus.STOPPING) {

            // 更新db中任务状态
            execution.setStartTime(new Date());
            updateStatus(execution, BatchStatus.STARTED);
            // 回调所有listener的beforeJob方法
            listener.beforeJob(execution);

            try {
                doExecute(execution);
            } catch (RepeatException e) {
                throw e.getCause(); // 搞不懂这里包一个RepeatException 有啥用
            }
        } else {
            // 任务状态时BatchStatus.STOPPING,说明任务已经停止,直接改成STOPPED
            // The job was already stopped before we even got this far. Deal
            // with it in the same way as any other interruption.
            execution.setStatus(BatchStatus.STOPPED);
            execution.setExitStatus(ExitStatus.COMPLETED);
        }

    } catch (JobInterruptedException e) {
        // 任务被打断 STOPPED
        execution.setExitStatus(getDefaultExitStatusForFailure(e, execution));
        execution.setStatus(BatchStatus.max(BatchStatus.STOPPED, e.getStatus()));
        execution.addFailureException(e);
    } catch (Throwable t) {
        // 其他原因失败 FAILED
        logger.error("Encountered fatal error executing job", t);
        execution.setExitStatus(getDefaultExitStatusForFailure(t, execution));
        execution.setStatus(BatchStatus.FAILED);
        execution.addFailureException(t);
    } finally {
        try {
            if (execution.getStatus().isLessThanOrEqualTo(BatchStatus.STOPPED)
                    && execution.getStepExecutions().isEmpty()) {
                ExitStatus exitStatus = execution.getExitStatus();
                ExitStatus newExitStatus =
                        ExitStatus.NOOP.addExitDescription("All steps already completed or no steps configured for this job.");
                execution.setExitStatus(exitStatus.and(newExitStatus));
            }

            // 计时器 计算总耗时
            timerSample.stop(BatchMetrics.createTimer("job", "Job duration",
                    Tag.of("name", execution.getJobInstance().getJobName()),
                    Tag.of("status", execution.getExitStatus().getExitCode())
            ));
            longTaskTimerSample.stop();
            execution.setEndTime(new Date());

            try {
                // 回调所有listener的afterJob方法  调用失败也不影响任务完成
                listener.afterJob(execution);
            } catch (Exception e) {
                logger.error("Exception encountered in afterJob callback", e);
            }
            // 写入db
            jobRepository.update(execution);
        } finally {
            // 释放控制
            JobSynchronizationManager.release();
        }

    }

}

3.2何时调用Reader?

在SimpleChunkProvider#provide中会分次调用reader,并将结果包装为Chunk返回。


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

  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 inputs = (Chunk) 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 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 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 transform(StepContribution contribution, Chunk inputs) throws Exception {
		Chunk outputs = new Chunk();
		for (Chunk.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 execute(TransactionCallback 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()));
}
  1. 信号量控制,在每个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);
}

文章来源于互联网:源码解析springbatch的job是如何运行的?

THE END
分享
二维码