我正在开发一个项目，我们使用 Spring Boot、Spring Batch 和 Camel。

批处理是通过调用 rest 端点启动的。 rest Controller 启动一个 camel 路由，启动 spring batch 作业流(通过 spring batch camel 组件)。

我无法控制调用我的应用程序的外部应用程序。我的申请是更大的夜间工作流程的一部分。

批处理作业可能需要很长时间才能完成，因此外部应用程序会通过另一个休息端点定期轮询我的批处理作业，询问作业是否已完成。它通过使用它想要状态的 jobExecution 的 id 轮询状态 rest 端点来实现这一点。

为了完成这个流程，我实现了一个休息 Controller ，它通过 ProducerTemplate 启动 Camel 路线。我的问题是在开始 Camel 路线后立即返回作业执行 ID。我不希望 rest 调用等到作业完成后再返回。

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

我尝试过使用异步调用和 future ，但没有成功。我也曾尝试使用 Camels 窃听无济于事。问题是只有“onComplete”事件。我需要一个在作业创建后立即返回但不运行的 Hook 。

例如，以下代码等待批处理作业完成，然后返回我要发回的 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; 
    } 
 
}     

Camel 路线是对 spring-batch-component 的简单调用

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 数据可用时立即返回它有什么想法吗？

请您参考如下方法:

不确定如何在 Camel 中执行此操作，但这里是使用 spring-rest 执行作业的示例。

@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 使 Job 异步。

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

引用documentation了解更多信息