How to reject new tasks in newSingleThreadExecutor when another task is running

Question

The actual scenario is Executors.newSingleThreadExecutor instance is executing a long-running task, I need to reject the new requests for this task until the completion of an existing one. while rejecting new requests, I need to simply send a message called "Thread is already running a task! Please wait until it completes".

Is it possible to implement using the newSingleThreadExecutor? Can Anyone Please help me...

Answer 1

The factory Executors.newSingleThreadExecutor() returns the equivalent to new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()), except that it is wrapped in another ExecutorService to prevent other code from casting it to ThreadPoolExecutor and changing the configuration. The wrapper also adds finalization support which you should not rely on anyway.

So you can construct a similar executor and alter its setup to your needs.

ExecutorService es = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
    new SynchronousQueue<>());

A SynchronousQueue has no capacity but can only hand elements over to already waiting consumers, i.e. will accept a Runnable only when there is already an idle worker thread. When the queue rejects the new job, the ThreadPoolExecutor will check its configured thread count and either, start a worker thread (at most one here) or call into a RejectedExecutionHandler. The default handler does throw a RejectedExecutionException, so we’re basically done with your requirements here.

The finetuning we can do, is to change the message of the RejectedExecutionException by providing our own RejectedExecutionHandler:

ExecutorService es = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
    new SynchronousQueue<>(),
    (runnable, executor) -> { throw new RejectedExecutionException(
        "Thread is already running a task! Please wait until it completes");
    });

When the amount of code having access to the ExecutorService is rather small and you can trust it to not doing things like casting es back to ThreadPoolExecutor and mess around, you can keep it this way. Otherwise, you can protect it against such modifications by wrapping it:

ExecutorService es = Executors.unconfigurableExecutorService(
    new ThreadPoolExecutor(1, 1, 200L, TimeUnit.MILLISECONDS, new SynchronousQueue<>(),
    (runnable, executor) -> { throw new RejectedExecutionException(
        "Thread is already running a task! Please wait until it completes");
    }));

Now we are as close as we can get to the behavior of Executors.newSingleThreadExecutor() plus your deviations, without implementing our own executor. The only thing this executor doesn’t have, is finalization support, but as this bug report suggests, it’s not a good idea to have it anyway. Take care to invoke shutdown() on it at the end of its lifetime (unless it overlaps with a call to System.exit(…) anyway).

The behavior can be tested with a program like

Future<?> previous = null;
for(int i = 0; i < 100; i++) {
    int jobID = i;
    System.out.println(jobID + " Trying to submit job");
    try {
        Future<?> next = es.submit(() -> {
            Thread.sleep(200);
            System.out.println("job " + jobID);
            return null;
        });
        if(previous != null && !previous.isDone()) {
            throw new AssertionError("new job accepted before previous completed");
        }
        previous = next;
    } catch(RejectedExecutionException ex) {
        System.out.println("rejected: " + ex);
    }
    Thread.sleep(ThreadLocalRandom.current().nextInt(90, 190));
}

This program attempts to submit multiple jobs and will throw an AssertionError if a job is accepted while the previous is not completed.