Настройка тайм-аута ThreadPoolExecutor

Question

Я хочу установить таймаут для моего ThreadPoolExecutor.

Я знаю, что я могу использовать

future.get(THREAD_TIMEOUT_MS, TimeUnit.MILLISECONDS);

Но этот код блокирует.

Чего я хочу добиться, так это того, что я могу создать несколько Runables, которые обрабатываются, например, с пулом из 4 потоков.

Если обработка потока занимает более 5 секунд, я хочу выдать исключение тайм-аута.

Это моя текущая настройка:

public class ServerExecutorService {

    public static int QUEUE_KEEP_ALIVE_TIME_SECONDS = 5;
    public static int MAXIMUM_POOL_SIZE = 12;
    public static int CORE_POOL_SIZE = 12;
    public static Logger LOG = LoggerFactory.getLogger(ServerExecutorService .class);
    public static int THREAD_TIMEOUT_MS = 5000;
    private LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(2);
    private RejectedHandler rejectedHandler = new RejectedHandler();


    private ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, QUEUE_KEEP_ALIVE_TIME_SECONDS, TimeUnit.SECONDS, linkedBlockingQueue);

    public ServerExecutorService () {
        executor.setRejectedExecutionHandler(this.rejectedHandler);
    }

    public void setRejectedHandler(RejectedHandler rejectedHandler) {
        executor.setRejectedExecutionHandler(rejectedHandler);
    }


    public void execute(Runnable runnable){
//        executor.execute(runnable);
//        Future<?> future = executor.submit(runnable);
        Future<?> future = executor.submit(runnable);

        try {
            future.get(THREAD_TIMEOUT_MS, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            System.out.println("Thread processing timeout.");
            LOG.warn("Thread processing timeout.", e);
        } catch (Exception e) {
            System.out.println("Thread processing error within ServerExecutorService ");
            LOG.error("Thread processing error within ServerExecutorService ", e);
        }
    }

}

Но, как вы можете видеть future.get(THREAD_TIMEOUT_MS, TimeUnit.MILLISECONDS); будет ждать, пока поток не закончится. Таким образом, следующие темы не запускаются.

Тестовое задание:

@Test
    public void testThreadPoolExhausted() {
        serverExecutorService.setRejectedHandler(rejectedHandler);
        for (int i = 0; i < 4; i++) {
            final int finalI = i;
            serverExecutorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("do something" + finalI);
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

В этом тесте второй поток запускается через 3 секунды, а не сразу.

Answer 1

Вы можете создать два потоковых пула, один для тайм-аутов и один для реальной работы.

ExecutorService timeoutService = Executors.newCachedThreadPool();
ExecutorService workerService = Executors.newCachedThreadPool();

public Future<?> submit(final Runnable runnable){
    return timeoutService.submit(() -> {
        try {
            Future<?> future = workerService.submit(runnable);
            future.get(100, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    });
}