完了する前にスプリングバッチジョブのジョブ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; 
    } 

}  

ラクダのルートは、私は、すぐにそれが利用可能であるよう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); 

    } 
} 

任意のアイデアですか？

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; 
    } 

は