RejectedExecutionException记一次job跑批失败

ERROR c.l.ehr.personnel.batch.job.BaseJob - 执行任务出错Exception
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@7f8991b8 rejected from java.util.concurrent.ThreadPoolExecutor@1665effd[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 433]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047) ~[na:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823) ~[na:1.8.0_111]
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369) ~[na:1.8.0_111]
at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112) ~[na:1.8.0_111]
at com.lianjia.ehr.personnel.batch.job.index.EmpIndexForOSJob.run(EmpIndexForOSJob.java:94) ~[batch-1.0-SNAPSHOT.jar:na]
at com.lianjia.ehr.personnel.batch.job.BaseJob.execute(BaseJob.java:58) ~[batch-1.0-SNAPSHOT.jar:na]
at org.quartz.core.JobRunShell.run(JobRunShell.java:202) [quartz-2.2.1.jar:na]
at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:573) [quartz-2.2.1.jar:na]

一次线上跑批失败看了下代码

@QuartzJob(name = "empIndexForOSJob", cronExp = "0 30 2 * * ? *", description = "为OpenSearch服务写empIndex内容")
public class EmpIndexForOSJob extends BaseJob {

    @Autowired
    private RunDataForOpenSearchService runDataForOpenSearchService;

    @Autowired
    private EmployeeService employeeService;
    @Autowired
    private PartTimeEmpService partTimeEmpService;

    private final static int threadCount = 10;

    private Logger logger = LoggerFactory.getLogger(EmpIndexForOSJob.class);

    private ExecutorService executor = Executors
            .newFixedThreadPool(threadCount,
                    new ThreadFactoryBuilder().setNameFormat("listen-empIndexForOSJob-thread-%d")
                            .setUncaughtExceptionHandler((t, e) -> {
                                logger.error("empIndexForOSJob process Throw Exception!", e);
                            }).build());

    @Override
    protected void run(JobExecutionContext context) throws Exception {

        logger.info("EmpIndexForOSJob start!");
        long start = System.nanoTime();
        // 遍历 emp
        List<Long> empIdList = Lists.newArrayList();
        empIdList.addAll(employeeService.queryAllEmpId());    //查出全部人员id集合
        logger.info("EmpIndexForOSJob running! [empIdList.size = {}]", empIdList.size());
        List<Range<Integer>> rangeList = DivideListUtil.divide(empIdList.size(), 1000);
        CountDownLatch countDownLatch = new CountDownLatch(rangeList.size());

        for (Range<Integer> range : rangeList) {
            logger.info("EmpIndexForOSJob range[{},{}]!", range.lowerEndpoint(), range.upperEndpoint());
            List<Long> subEmpIdList = empIdList.subList(range.lowerEndpoint(), range.upperEndpoint());
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        for (Long empId : subEmpIdList) {
                            try {
                                runDataForOpenSearchService.execute(empId, "employee", ActionType.UPDATE);
                            } catch (Throwable throwable) {
                                logger.error("empIndexForOSJob process Throw Exception! [empId={}]", empId);
                            }
                        }
                    } catch (Throwable e) {
                        logger.error("empIndexForOSJob process Throw Exception! [e={}]", e);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
            });
        }
        countDownLatch.await();
        executor.shutdown();
        // 遍历 parttimeEmp
        List<Long> partTimeEmpIdList = Lists.newArrayList();
        partTimeEmpIdList.addAll(partTimeEmpService.queryAllPartTimeEmpId());  //查出全部相关人员id集合
        logger.info("EmpIndexForOSJob running! [partTimeEmpIdList.size = {}]", partTimeEmpIdList.size());
        List<Range<Integer>> rangeParttimeList = DivideListUtil.divide(partTimeEmpIdList.size(), 1000);
        CountDownLatch parttimeCountDownLatch = new CountDownLatch(rangeParttimeList.size());
        for (Range<Integer> range : rangeParttimeList) {
            logger.info("EmpIndexForOSJob range[{},{}]!", range.lowerEndpoint(), range.upperEndpoint());
            List<Long> subEmpIdList = partTimeEmpIdList.subList(range.lowerEndpoint(), range.upperEndpoint());
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        for (Long empId : subEmpIdList) {
                            try {
                                runDataForOpenSearchService.execute(empId, "parttime_emp", ActionType.UPDATE);
                            } catch (Throwable throwable) {
                                logger.error("empIndexForOSJob process Throw Exception! [parttimeEmpId={}]", empId);
                            }
                        }
                    } catch (Throwable e) {
                        logger.error("empIndexForOSJob process Throw Exception! [e={}]", e);
                    } finally {
                        parttimeCountDownLatch.countDown();
                    }
                }
            });
        }

        parttimeCountDownLatch.await();
        executor.shutdown();
        long end = System.nanoTime();
        logger.info("EmpIndexForOSJob end! costs={}ms", (end - start) / 1000000);
    }

}

 分析：

看了下报错log，RejectedExecutionException 怀疑和线程池有关系

原来问题出现在第58行 executor.shutdown();

线程池的shutdown()方法：

当线程池调用该方法时,线程池的状态则立刻变成SHUTDOWN状态。此时，则不能再往线程池中添加任何任务，否则将会抛出RejectedExecutionException异常。但是，此时线程池不会立刻退出，直到添加到线程池中的任务都已经处理完成，才会退出。

线程池的shutdownNow()方法：
根据JDK文档描述，大致意思是：执行该方法，线程池的状态立刻变成STOP状态，并试图停止所有正在执行的线程，不再处理还在池队列中等待的任务，当然，它会返回那些未执行的任务。
它试图终止线程的方法是通过调用Thread.interrupt()方法来实现的，但是大家知道，这种方法的作用有限，如果线程中没有sleep 、wait、Condition、定时锁等应用, interrupt()方法是无法中断当前的线程的。所以，ShutdownNow()并不代表线程池就一定立即就能退出，它可能必须要等待所有正在执行的任务都执行完成了才能退出。

最后：因为后面还有任务要添加不要中途shutdown()，把第58行代码删掉即可，等所有任务都执行完毕在执行shutdown()。