我想实现重新启动作业的功能，以便从初始sage启动作业。我面临两个问题。
第一个问题：当我第一次重新启动作业时，它会创建一个新的作业实例id，并且表现得像一个新作业。第二次，它将重新启动并使用相同的作业实例id运行。(我从rest控制器发送了执行id(
第二个问题：当我重新启动它时，它将从初始阶段开始。
自定义阅读器：
package com.orange.alc.dabekdataload.reader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.AfterStep;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Scope;
import org.springframework.context.annotation.ScopedProxyMode;
import org.springframework.core.io.FileSystemResource;
import org.springframework.stereotype.Component;
import com.orange.alc.dabekdataload.constants.PostalHeader;
import com.orange.alc.dabekdataload.dto.PostalDto;
@Component("itemReader")
@Scope(value = "step", proxyMode = ScopedProxyMode.TARGET_CLASS)
public class PostalReader implements ItemReader<PostalDto>, ItemStream{
private static final Logger LOGGER = LoggerFactory.getLogger(PostalReader.class);
@Value("#{jobParameters[fullPathFileName]}")
public String fileName;
private int currentIndex = 0;
private static final String CURRENT_INDEX = "current.index";
private FlatFileItemReader<PostalDto> reader;
@BeforeStep
public void beforeStep(StepExecution stepExecution) {
LOGGER.info("Executing batch reader...");
reader = new FlatFileItemReader<>();
reader.setResource(new FileSystemResource(fileName));
reader.setLinesToSkip(1);
reader.setLineMapper(new DefaultLineMapper<PostalDto>() {{
setLineTokenizer(new DelimitedLineTokenizer() {{
setNames(PostalHeader.getPostalColumnNames());
}});
setFieldSetMapper(new PostalFieldSetMapper());
}});
reader.setSaveState(true);
reader.open(stepExecution.getExecutionContext());
}
@Override
public PostalDto read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
reader.setCurrentItemCount(currentIndex++);
return reader.read();
}
@AfterStep
public void afterStep(StepExecution stepExecution) {
LOGGER.info("Closing the reader...");
reader.close();
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
if(executionContext.containsKey(CURRENT_INDEX)){
currentIndex = new Long(executionContext.getLong(CURRENT_INDEX)).intValue();
} else{
currentIndex = 0;
}
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putLong(CURRENT_INDEX, new Long(currentIndex).longValue());
}
@Override
public void close() throws ItemStreamException {
}
}
作业重新启动代码：
@Override
public void restartJob(Long jobId) throws JobInstanceAlreadyCompleteException, NoSuchJobExecutionException, NoSuchJobException, JobRestartException, JobParametersInvalidException {
LOGGER.info("Restarting job with JobId: {}", jobId);
jobOperator.restart(jobId);
}
如果你需要我这边的任何代码，请告诉我。
自定义读卡器(PostalReader(中使用的委托读卡器"FlatFileItemReader"未遵守ItemStream合同。您需要在项目读取器的相应open/update/close方法中的委托读取器上调用open/update/close。类似于：
public class PostalReader implements ItemReader<PostalDto>, ItemStream{
private FlatFileItemReader<PostalDto> reader;
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
reader.open(executionContext);
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
reader.update(executionContext);
}
@Override
public void close() throws ItemStreamException {
reader.close();
}
}
itemStream的update方法将把exuctionContext放在当前Step的执行上下文中。但是，当您重新启动作业时，该步骤将作为新步骤重新运行，并且无法获取上一步骤的执行上下文。为了从数据重新启动的位置开始处理数据，您需要将状态保存在作业执行上下文中。