Java多线程监听K8s Job运行状态
Java多线程监听K8s Job的状态,如果在项目中需要知道Job运行状态,可以利用线程池启动线程进行检测,完成后推送消息,然后关闭线程;1,配置线程池@Configuration@EnableAsyncpublic class TaskPoolConfig {@Bean("taskExecutor")public Executor taskExecutro(){ThreadPoolTaskExe
·
Java多线程监听K8s Job的状态,如果在项目中需要知道Job运行状态,可以利用线程池启动线程进行检测,完成后推送消息,然后关闭线程;
1,配置线程池
@Configuration
@EnableAsync
public class TaskPoolConfig {
@Bean("taskExecutor")
public Executor taskExecutro(){
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(10);
taskExecutor.setMaxPoolSize(50);
taskExecutor.setQueueCapacity(200);
taskExecutor.setKeepAliveSeconds(60);
taskExecutor.setThreadNamePrefix("taskExecutor--");
taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
taskExecutor.setAwaitTerminationSeconds(60);
return taskExecutor;
}
}
2,线程类
/**
* Job监听线程,当评测任务结束关闭线程
*/
@Slf4j
@Component
public class JobStateMonitorThread {
@Autowired
private static EvaluationService evaluationService;
@Autowired
private static RedisUtil redisUtil;
@Resource
public void setEvaluationService(EvaluationService evaluationService) {
JobStateMonitorThread.evaluationService = evaluationService;
}
@Resource
public void setRedisUtil(RedisUtil redisUtil) {
JobStateMonitorThread.redisUtil = redisUtil;
}
@Async("taskExecutor")
public void tesTask(String no,String jobName) throws IOException, InterruptedException, ApiException {
log.info("Job:{}监控线程启动{}" ,jobName,Thread.currentThread().getName());
Thread.sleep(10000);
boolean jobActive = true;
while (jobActive) {
String state = K8sUtil.getJobState(jobName);
log.info("获取【{}】运行状态时间{},运行状态:{}" ,jobName,new Date(),state);
if (state.equals("running")){
log.info("{}运行中" ,jobName);
Thread.sleep(60000);
}else {
log.info("{}运行状态为完成" ,jobName);
if (state.equals("success")){
//运行成功
//放到队列
redisUtil.pub("JOB_START",no);
}else {
//运行失败
log.info("{}运行状态为失败,推送消息" ,no);
evaluationService.addResult(no);
}
log.info("{}线程关闭" ,jobName);
jobActive = false;
Thread.currentThread().interrupt();
throw new RuntimeException("Job任务运行完成,异常法强制关闭当前监听线程!");
}
}
}
public JobStateMonitorThread(){
}
}
3,调用线程
//启动线程监听
log.info("启动线程监听-{}",jobName);
JobStateMonitorThread jobStateMonitorThread = new JobStateMonitorThread();
jobStateMonitorThread.tesTask(no,jobName);
更多推荐
已为社区贡献3条内容
所有评论(0)