在完成彈簧批處理作業之前返回作業ID「立即」

Question

我正在使用Spring Boot，Spring Batch和Camel的項目。

批處理過程是通過調用其餘端點來啓動的。其餘的控制器啓動一個駱駝路線，啓動彈簧批處理作業流程（通過彈簧批量駱駝組件）。

我無法控制調用我的應用程序的外部應用程序。我的應用程序是一個更大的夜間工作流程的一部分。

批處理作業可能需要很長時間才能完成，因此外部應用程序會定期通過另一個休息端點輪詢我的批處理作業，詢問作業是否完成。它通過輪詢一個狀態休息端點並使用jobExecution的id來執行此操作。

爲了完成這個流程，我實現了一個通過ProducerTemplate啓動駱駝路線的休息控制器。我的問題是在開始駱駝路線後立即返回作業執行ID。我不希望剩下的電話等到作業完成後才能返回。

startJobViaRestCall ------> createBatchJob ----> runBatchJobUntilDone 
            | 
            | 
     Return jobExecutionData | 
<---------------------------------- 

我試過使用異步調用和期貨，但沒有運氣。我也試圖用駱駝竊聽無濟於事。問題是隻有「onComplete」事件。我需要一個在作業創建完成後立即返回的鉤子，但不能運行。

例如，下面的代碼會一直等到批量作業完成，然後再返回我想要發回的JobExecution數據（如json）。這是有道理的，因爲extractFutureBody將等待響應準備就緒。

@RestController 
@Slf4j 
public class BatchJobController { 

    @Autowired 
    ProducerTemplate producerTemplate; 

    @RequestMapping(value = "/batch/job/start", method = RequestMethod.GET) 
    @ResponseBody 
    public String startBatchJob() { 
     log.info("BatchJob start called..."); 

     String jobExecution = producerTemplate.extractFutureBody(producerTemplate.asyncRequestBody(BatchRoute.ENDPOINT_JOB_START, ""), String.class); 

     return jobExecution; 
    } 

}  

駱駝的路線是春季批次組分

public class BatchRoute<I, O> extends BaseRoute { 

    private static final String ROUTE_START_BATCH = "spring-batch:springBatchJob"; 

    @Override 
    public void configure() { 

     super.configure(); 
     from(ENDPOINT_JOB_START).to(ROUTE_START_BATCH); 

    } 
} 

任何想法，我怎麼能儘快它是可用返回JobExecution數據的簡單通話？

Answer 1

不知道如何在駱駝中做到這一點，但這裏是使用春天休息的示例作業執行。

@RestController 
public class KpRest { 

    private static final Logger LOG = LoggerFactory.getLogger(KpRest.class); 
    private static String RUN_ID_KEY = "run.id"; 

    @Autowired 
    private JobLauncher launcher; 

    private final AtomicLong incrementer = new AtomicLong(); 


    @Autowired 
    private Job job; 


    @RequestMapping("/hello") 
    public String sayHello(){ 

     try { 
      JobParameters parameters = new JobParametersBuilder().addLong(RUN_ID_KEY, incrementer.incrementAndGet()).toJobParameters(); 
      JobExecution execution = launcher.run(job, parameters); 
      LOG.info("JobId {}, JobStatus {}", execution.getJobId(), execution.getStatus().getBatchStatus()); 
      return String.valueOf(execution.getJobId()); 
     } catch (JobExecutionAlreadyRunningException | JobRestartException | JobInstanceAlreadyCompleteException 
       | JobParametersInvalidException e) { 
      LOG.info("Job execution failed, {}", e); 
     } 
     return "Some Error"; 
    } 
} 

您可以通過修改JobLauncher來使作業異步。

@Bean 
    public JobLauncher simpleJobLauncher(JobRepository jobRepository){ 
     SimpleJobLauncher jobLauncher = new SimpleJobLauncher(); 
     jobLauncher.setJobRepository(jobRepository); 
     jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor()); 
     return jobLauncher; 
    } 

請參考documentation更多信息