How to reschedule a task using a ScheduledExecutorService?

Question

I saw this in the java docs: ScheduledAtFixedRate, it says

If any execution of the task encounters an exception, subsequent executions are suppressed

I don't want this to happen in my application. Even if I see an exception I would always want the subsequent executions to occur and continue. How can I get this behavior from ScheduledExecutorService.

Answer 1

Surround the Callable.call method or the Runnable.run method with a try/catch...

eg:

public void run()
{
    try
    {
        // ... code
    }
    catch(final IOException ex)
    {
        // handle it
    }
    catch(final RuntimeException ex)
    {
        // handle it
    }
    catch(final Exception ex)
    {
        // handle it
    }
    catch(final Error ex)
    {
        // handle it
    }
    catch(final Throwable ex)
    {
        // handle it
    }
}

Note that catching anything other than what the compiler tells you too (the IOException in my sample) isn't a good idea, but there are some times, and this sounds like one of them, that it can work out if you handle it properly.

Remember that things like Error are very bad - the VM ran out of memory etc... so be careful how you handle them (which is why I separated them out into their own handlers rather than just doing catch(final Throwable ex) and nothing else).

Answer 2

Try VerboseRunnable class from jcabi-log, which does the wrapping suggested by TofuBeer:

import com.jcabi.log.VerboseRunnable;
Runnable runnable = new VerboseRunnable(
  Runnable() {
    public void run() { 
      // do business logic, may Exception occurs
    }
  },
  true // it means that all exceptions will be swallowed and logged
);

Now, when anybody calls runnable.run() no exceptions are thrown. Instead, they are swallowed and logged (to SLF4J).

Answer 3

I had the same problem. I also tried that try block within run() method but it doesn't work.

So I did something is working so far:

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class Test2 {

    static final ExecutorService pool = Executors.newFixedThreadPool(3);

    static final R1 r1 = new R1();
    static final R2 r2 = new R2();

    static final BlockingQueue deadRunnablesQueue = new LinkedBlockingQueue<IdentifiableRunnable>();

    static final Runnable supervisor = new Supervisor(pool, deadRunnablesQueue);

    public static void main(String[] args) {
        pool.submit(r1);
        pool.submit(r2);
        new Thread(supervisor).start();
    }

    static void reSubmit(IdentifiableRunnable r) {
        System.out.println("given to an error, runnable [" + r.getId()
                + "] will be resubmited");
        deadRunnablesQueue.add(r);
    }

    static interface IdentifiableRunnable extends Runnable {
        String getId();
    }

    static class Supervisor implements Runnable {
        private final ExecutorService pool;
        private final BlockingQueue<IdentifiableRunnable> deadRunnablesQueue;

        Supervisor(final ExecutorService pool,
                final BlockingQueue<IdentifiableRunnable> deadRunnablesQueue) {
            this.pool = pool;
            this.deadRunnablesQueue = deadRunnablesQueue;
        }

        @Override
        public void run() {
            while (true) {
                IdentifiableRunnable r = null;
                System.out.println("");
                System.out
                        .println("Supervisor will wait for a new runnable in order to resubmit it...");
                try {
                    System.out.println();
                    r = deadRunnablesQueue.take();
                } catch (InterruptedException e) {
                }
                if (r != null) {
                    System.out.println("Supervisor got runnable [" + r.getId()
                            + "] to resubmit ");
                    pool.submit(r);
                }
            }
        }
    }

    static class R1 implements IdentifiableRunnable {
        private final String id = "R1";
        private long l;

        @Override
        public void run() {
            while (true) {
                System.out.println("R1 " + (l++));
                try {
                    Thread.currentThread().sleep(5000);
                } catch (InterruptedException e) {
                    System.err.println("R1 InterruptedException:");
                }
            }
        }

        public String getId() {
            return id;
        }
    }

    static class R2 implements IdentifiableRunnable {
        private final String id = "R2";
        private long l;

        @Override
        public void run() {
            try {
                while (true) {
                    System.out.println("R2 " + (l++));
                    try {
                        Thread.currentThread().sleep(5000);
                    } catch (InterruptedException e) {
                        System.err.println("R2 InterruptedException:");
                    }
                    if (l == 3) {
                        throw new RuntimeException(
                                "R2 error.. Should I continue to process ? ");
                    }
                }
            } catch (final Throwable t) {
                t.printStackTrace();
                Test2.reSubmit(this);
            }
        }

        public String getId() {
            return id;
        }
    }

}

You can try to comment out Test2.reSubmit(this) to see that without it, R2 will stop working.