java - 我无法使用 ExecutorService 获得 RejectedExecutionException

Question

我创建调度程序来测试处理 RejectedExecutionException:

@Component
public class TestScheduler {

    private final TestService testService;
    private ExecutorService executorService;

    public TestScheduler(TestService testService) {
        this.testService = testService;
    }

    @PostConstruct
    public void init() {
        executorService = Executors.newFixedThreadPool(5);
    }

    @Scheduled(fixedRate = 10L)
    public void test() {
        System.out.println("test");
        executorService.execute(testService::print);
    }
}

延迟 70 秒的服务:

@Component
public class TestService {

    public void print() {
        System.out.println("print start");
        try {
            TimeUnit.SECONDS.sleep(70);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("print end");
    }
}

我等待下一个逻辑:

1. 调度程序调用 executorService.execute(testService::print) 5 次
2. 每个 testService::print 将执行 70 秒
3. 当execute方法第六次调用时，我得到RejectedExecutionException

但我没有得到异常。我有这个日志:

test
print start
2018-10-22 11:26:45.543  INFO 5960 --- [           main] c.e.s.SchedullerExceptionsApplication    : Started SchedullerExceptionsApplication in 0.661 seconds (JVM running for 1.108)
test
print start
test
print start
test
print start
test
print start
test
...
70 seconds print test

编辑

在实际项目中我有这样的代码:

@PostConstruct
    public void init() {
        executorService = Executors.newFixedThreadPool(100, new CustomizableThreadFactory("SendRequestExecutor-"));
    }

@Scheduled(fixedDelay = 1000L)
public void sendReady() {
    try {
        List<Message> messages = messageService.findReadyToSend();
        for (Message message : messages) {
            message.setStatus(IN_PROCESS);
            Message savedMessage = messageService.save(message);
            executorService.execute(() -> sendRequestService.send(savedMessage.getGuid()));
        }
    } catch (Exception e) {
        log.error("handle: " + e.getMessage());
    }
}

这是否意味着这段代码是错误的？因为这种情况可能会发生，所以我将实体更改为状态 IN_PROCESS 并在尝试执行时 - 如果 executorService 已满，我不会收到异常，并且 executorService 不会执行我的任务？

Answer 1

定义执行器时涉及两个方面。

1. 执行程序将使用的线程数。这会限制执行程序可以运行的并发任务数。这就是您通过 Executors.newFixedThreadPool(5) 设置的内容。
2. 底层任务提交队列的大小。这会限制底层队列可以存储的任务数量，直到抛出异常。 newFixedThreadPool 创建的执行程序使用无界队列，因此您不会遇到异常。

您可以通过如下方式创建自己的执行程序服务并向其提交 11 个任务(5 个用于使用所有线程，5 个用于填充底层任务队列，1 个用于溢出)来实现所需的行为。

new ThreadPoolExecutor(5, 
                       5, 
                       2000L, 
                       TimeUnit.MILLISECONDS, 
                       new ArrayBlockingQueue<Runnable>(5, true), 
                       new ThreadPoolExecutor.CallerRunsPolicy());