Java safely store and restore unfinished tasks in a thread pool

Question

I have a fixed thread pool in my java web application.

 App.aes =  Executors.newFixedThreadPool(3);

It was used for executing asynchronous tasks. Those tasks can take hours to finish. So, if I need to do a application reloading, I need to check if there are asynchronous tasks running, if yes I need to store those tasks in the waiting queue to somewhere and restore them after a application reloading.

I did some test:

public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(1);
        for(int i=0; i< 5; i++){
            final int c = i + 1;
            es.submit(new Runnable(){
                @Override
                public void run() {
                    System.out.println("current running " + c);
                    try {
                        Thread.sleep(10000);    // 10 sec
                    } catch (InterruptedException e) {
                        System.out.println("interrupted  " + c);
                    }
                }

            });
        }

        Thread.sleep(15000);
        List<Runnable> rems = es.shutdownNow();
        System.out.println("Remaining " + rems.size());
        System.out.println("--------- restore remaining task ----------");

        es = Executors.newFixedThreadPool(1);
        for(Runnable r : rems){
            es.submit(r);
        }
    }

the output is:

current running 1
current running 2
interrupted  2
Remaining 4
--------- restore remaining task ----------
current running 3
current running 4
current running 5
current running 6

And this is not the result I am looking for. The interrupted tasks will not be recovered. And the API docs proved this:

Attempts to stop all actively executing tasks, halts the processing of waiting tasks, and returns a list of the tasks that were awaiting execution. 

how can we safely store and restore unfinished tasks in a java thread pool? My real task is fail-safe which means each task can be rerun again and again. And the order of being executed does not matter.

My web application is deployed on weblogic, and the thread pool is started up by a servlet, a ServletContextListener is registerd to shutdown the thread pool.

---

I am expecting two options:

• option 1. No need to interrupt active tasks, wait them finish, and then save all those waiting tasks then shutdown thread pool.

  pros: no need to worry about any unpredictable condition caused by interruption.
  cons: this will need to wait all actively running task to be finished. depend on thread pool size and time cost of each task, the waiting time can be long.

• option 2. Interrupt active task, save all unfinished tasks then
  shutdown thread pool.

option 1 is the idea solution for me.

Answer 1

How about creating a list of interrupted tasks? You have already the catch enabling you to execute code for a specific interrupted thread:

public static void main(String[] args) throws InterruptedException {
    ExecutorService es = Executors.newFixedThreadPool(1);
    List<Runnable> interruptedTasks = new CopyOnWriteArrayList<Runnable>(); //create the list of interrupted threads/tasks
        //Edited - after the right comment by @Luke Lee (AbstractList cannot be instantiated and the operation in the catch block should be thread-safe)
    for(int i=0; i< 5; i++){
        final int c = i + 1;
        es.submit(new Runnable(){
            @Override
            public void run() {
                System.out.println("current running " + c);
                try {
                    Thread.sleep(10000);    // 10 sec
                } catch (InterruptedException e) {
                    System.out.println("interrupted  " + c);
                    interruptedTasks.add(this); //add this interrupted instance to the list
                }
            }

        });
    }

    Thread.sleep(15000);
    List<Runnable> rems = es.shutdownNow();
    System.out.println("Remaining " + rems.size());
    System.out.println("--------- restore remaining task ----------");

    es = Executors.newFixedThreadPool(1);
    for (Runnable r : interruptedTasks){ //add the interrupted Runnables to the new pool
        es.submit(r);
    }
    for(Runnable r : rems){
        es.submit(r);
    }
}

I didn't test this - but this should work, I believe. If you need some sort of clearing up before rerunning the task you can of course do them in the catch while adding the instance to the interrupted tasks list.

Obviously this helps the simplified test - but you can do the same in a more complex design: you can have the Runnable as a class that gets the pool's object as a parameter in the constructor rather than using an anonymous class. Only after creating the instance you submit it to the pool. That way it has the ability to add itself to the pool if interrupted. But then you need to add a setPool(ExecutorSerive) method to the Runnable and call it, resetting the pool's object to the new one, before rerunning it (in the for-loop that I added, just before the submit line).

---

Edit: Just saw your edit - about the options. My suggestion was obviously the 2nd option. For the first option I think you can use the example code in the documentation here, specifically - look at awaitTermination

The relevant section:

void shutdownAndAwaitTermination(ExecutorService pool) {
   pool.shutdown(); // Disable new tasks from being submitted
   try {
     // Wait a while for existing tasks to terminate
     if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
       pool.shutdownNow(); // Cancel currently executing tasks
       // Wait a while for tasks to respond to being cancelled
       if (!pool.awaitTermination(60, TimeUnit.SECONDS))
           System.err.println("Pool did not terminate");
     }
   } catch (InterruptedException ie) {
     // (Re-)Cancel if current thread also interrupted
     pool.shutdownNow();
     // Preserve interrupt status
     Thread.currentThread().interrupt();
   }
}