Java多线程Callable实用篇

1、线程池的配置查看CompletableFuture2、业务场景为只要有一任务失败则所有任务都失败注意线程池的拒绝策略为抛出异常。2.1、重写Callableclass MyCallable implements CallableThreadResult{ private final String params; private final ThreadResult threadResult; AtomicBoolean errorFlag;//标识如果有一个任务失败则其它任务不用执行 public MyCallable(String params,AtomicBoolean errorFlag,ThreadResult threadResult){ this.params params; this.threadResult threadResult; this.errorFlag errorFlag; } //这里将异常直接抛出方便查询问题。根据实际情况处理 Override public ThreadResult call() throws Exception { //1、其中有任务失败则不需要再执行 if(errorFlag.get()){ threadResult.setCode(1000); threadResult.setMessage(cancel task handel!); return threadResult; } if(ObjectUtils.isEmpty(params)){ threadResult.setCode(1001); threadResult.setErrorMessage(param cat not be null!); return threadResult; } //2、各自的Exception由具体的代码块处理这里抛出由调用者处理 String selectSql select * from springdemo; ListMapString,Object dataList null; dataList jdbcTemplate.queryForList(selectSql); Thread.sleep(1000); //测试场景模拟抛错 if(3000.equals(params)){ throw new RuntimeException(handle test : callable runtime exception!); } StringBuilder sb new StringBuilder(); if(!ObjectUtils.isEmpty(dataList)){ for(MapString,Object dataMap : dataList){ sb.append(number:).append(dataMap.get(number)).append(,name:).append(dataMap.get(name)); } } //标识执行成功 threadResult.setStatus(true); return threadResult; } }2.2、调用public String callableInvokeAllDemo() { IPrintUtil.print( begin: available queue : myThreadPool.getQueue().remainingCapacity()); int count 9000; //报错标识 AtomicBoolean errorFlag new AtomicBoolean(false); //新建任务 ListCallableThreadResult callableList new ArrayList(); for(int i0;icount;i){ ThreadResult result new ThreadResult(); //listResult.add(result); callableList.add(new MyCallable(i,errorFlag,result)); } //开始提交任务并执行 ListFutureThreadResult futures new ArrayList(); //第1关键点线程池提交任务时要处理线程池满的情况。如果是直接抛出异常 try{ for(CallableThreadResult callable : callableList) { FutureThreadResult futureResult myThreadPool.submit(callable); futures.add(futureResult); } } catch (RejectedExecutionException e){ //1、告诉执行任务不需要再执行了 errorFlag.set(true); //2、尝试停止正常执行的任务 for(FutureThreadResult f: futures) { if(!f.isDone()){ f.cancel(true); } } //3、业务处理根据实际情况处理。这里直接抛出异常 throw new RuntimeException(e.getMessage()); } //获取执行结果这里会有一个问题就是第一个任务结果执行很久则线程会卡住 String errorMessage ; for(FutureThreadResult future : futures) { try { if (future.isCancelled()) { //标识其它任务不需要再执行 errorFlag.set(true); IPrintUtil.print(Thread Id : Thread.currentThread().getId() isCancelled); //其它任务不需要再获取结果 break; } else { IPrintUtil.print(Thread Id : Thread.currentThread().getId() isDone); } //设置一个超时时间 ThreadResult res future.get(30,TimeUnit.MINUTES); //如果有执行失败则其它任务不需要再执行 if(!res.isStatus()) { errorFlag.set(true); errorMessage res.getMessage(); //其它任务不需要再获取结果 } } catch (InterruptedException e) { IPrintUtil.print(e.getMessage()); IPrintUtil.print(e1); //标识其它任务不需要再执行 errorFlag.set(true); errorMessage e.getMessage(); break; } catch (ExecutionException e) { IPrintUtil.print(e.getMessage()); IPrintUtil.print(e2); //标识其它任务不需要再执行 errorFlag.set(true); errorMessage e.getMessage(); break; } catch (Exception e){ IPrintUtil.print(Thread Id : Thread.currentThread().getId() exception 3 is : e.getMessage()); IPrintUtil.print(e3); //标识其它任务不需要再执行 errorFlag.set(true); errorMessage e.getMessage(); break; } } //任务执行失败 if(errorFlag.get()){ //尝试取消任务 for(FutureThreadResult future : futures) { if(!future.isDone()){ IPrintUtil.print(-------cancel task-------); future.cancel(true); } } //这里看具体情况处理 //throw new Exception(errorMessage); return error: errorMessage; } return success; }