java.util.concurrent.ScheduledExecutorService
Wraps the task's Runnable in a - * {@link org.springframework.scheduling.support.DelegatingExceptionProofRunnable} - * if necessary, according to the + * {@link org.springframework.scheduling.support.DelegatingErrorHandlingRunnable} + * that will catch and log the Exception. If necessary, it will suppress the + * Exception according to the * {@link #setContinueScheduledExecutionAfterException "continueScheduledExecutionAfterException"} * flag. * @param task the ScheduledExecutorTask to schedule * @return the actual Runnable to schedule (may be a decorator) */ protected Runnable getRunnableToSchedule(ScheduledExecutorTask task) { - boolean propagateException = !this.continueScheduledExecutionAfterException; - return new DelegatingExceptionProofRunnable(task.getRunnable(), propagateException); + return this.continueScheduledExecutionAfterException + ? new DelegatingErrorHandlingRunnable(task.getRunnable(), ErrorHandler.LOG_AND_SUPPRESS) + : new DelegatingErrorHandlingRunnable(task.getRunnable(), ErrorHandler.LOG_AND_PROPAGATE); } diff --git a/org.springframework.context/src/main/java/org/springframework/scheduling/concurrent/TaskUtils.java b/org.springframework.context/src/main/java/org/springframework/scheduling/concurrent/TaskUtils.java new file mode 100644 index 00000000000..c03792be83b --- /dev/null +++ b/org.springframework.context/src/main/java/org/springframework/scheduling/concurrent/TaskUtils.java @@ -0,0 +1,52 @@ +/* + * Copyright 2002-2009 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.scheduling.concurrent; + +import java.util.concurrent.Future; + +import org.springframework.scheduling.support.DelegatingErrorHandlingRunnable; +import org.springframework.scheduling.support.ErrorHandler; + +/** + * @author Mark Fisher + * @since 3.0 + */ +abstract class TaskUtils { + + /** + * Decorates the task for error handling. If the provided + * {@link ErrorHandler} is not null, it will be used. Otherwise, + * repeating tasks will have errors suppressed by default whereas + * one-shot tasks will have errors propagated by default since those + * errors may be expected through the returned {@link Future}. In both + * cases, the errors will be logged. + */ + static DelegatingErrorHandlingRunnable errorHandlingTask( + Runnable task, ErrorHandler errorHandler, boolean isRepeatingTask) { + + if (task instanceof DelegatingErrorHandlingRunnable) { + return (DelegatingErrorHandlingRunnable) task; + } + ErrorHandler eh = errorHandler != null ? errorHandler : getDefaultErrorHandler(isRepeatingTask); + return new DelegatingErrorHandlingRunnable(task, eh); + } + + static ErrorHandler getDefaultErrorHandler(boolean isRepeatingTask) { + return (isRepeatingTask ? ErrorHandler.LOG_AND_SUPPRESS : ErrorHandler.LOG_AND_PROPAGATE); + } + +} diff --git a/org.springframework.context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java b/org.springframework.context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java index c10146d82d1..42a01029350 100644 --- a/org.springframework.context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java +++ b/org.springframework.context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java @@ -33,6 +33,7 @@ import org.springframework.core.task.TaskRejectedException; import org.springframework.scheduling.SchedulingTaskExecutor; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.support.ErrorHandler; import org.springframework.util.Assert; /** @@ -40,16 +41,20 @@ import org.springframework.util.Assert; * a native {@link java.util.concurrent.ScheduledThreadPoolExecutor}. * * @author Juergen Hoeller + * @author Mark Fisher * @since 3.0 * @see #setPoolSize * @see #setThreadFactory + * @see #setErrorHandler */ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport implements TaskScheduler, SchedulingTaskExecutor { - private int poolSize = 1; + private volatile int poolSize = 1; - private ScheduledExecutorService scheduledExecutor; + private volatile ScheduledExecutorService scheduledExecutor; + + private volatile ErrorHandler errorHandler; /** @@ -61,6 +66,13 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport this.poolSize = poolSize; } + /** + * Provide an {@link ErrorHandler} strategy. + */ + public void setErrorHandler(ErrorHandler errorHandler) { + Assert.notNull(errorHandler, "'errorHandler' must not be null"); + this.errorHandler = errorHandler; + } protected ExecutorService initializeExecutor( ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) { @@ -102,7 +114,7 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport public void execute(Runnable task) { Executor executor = getScheduledExecutor(); try { - executor.execute(task); + executor.execute(errorHandlingTask(task, false)); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); @@ -116,7 +128,7 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport public Future submit(Runnable task) { ExecutorService executor = getScheduledExecutor(); try { - return executor.submit(task); + return executor.submit(errorHandlingTask(task, false)); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); @@ -126,6 +138,9 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport public Future submit(Callable task) { ExecutorService executor = getScheduledExecutor(); try { + if (this.errorHandler != null) { + task = new DelegatingErrorHandlingCallable(task, this.errorHandler); + } return executor.submit(task); } catch (RejectedExecutionException ex) { @@ -143,7 +158,9 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport public ScheduledFuture schedule(Runnable task, Trigger trigger) { ScheduledExecutorService executor = getScheduledExecutor(); try { - return new ReschedulingRunnable(task, trigger, executor).schedule(); + ErrorHandler errorHandler = this.errorHandler != null ? + this.errorHandler : TaskUtils.getDefaultErrorHandler(true); + return new ReschedulingRunnable(task, trigger, executor, errorHandler).schedule(); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); @@ -154,7 +171,7 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport ScheduledExecutorService executor = getScheduledExecutor(); long initialDelay = startTime.getTime() - System.currentTimeMillis(); try { - return executor.schedule(task, initialDelay, TimeUnit.MILLISECONDS); + return executor.schedule(errorHandlingTask(task, false), initialDelay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); @@ -165,7 +182,7 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport ScheduledExecutorService executor = getScheduledExecutor(); long initialDelay = startTime.getTime() - System.currentTimeMillis(); try { - return executor.scheduleAtFixedRate(task, initialDelay, period, TimeUnit.MILLISECONDS); + return executor.scheduleAtFixedRate(errorHandlingTask(task, true), initialDelay, period, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); @@ -175,7 +192,7 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport public ScheduledFuture scheduleAtFixedRate(Runnable task, long period) { ScheduledExecutorService executor = getScheduledExecutor(); try { - return executor.scheduleAtFixedRate(task, 0, period, TimeUnit.MILLISECONDS); + return executor.scheduleAtFixedRate(errorHandlingTask(task, true), 0, period, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); @@ -186,7 +203,7 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport ScheduledExecutorService executor = getScheduledExecutor(); long initialDelay = startTime.getTime() - System.currentTimeMillis(); try { - return executor.scheduleWithFixedDelay(task, initialDelay, delay, TimeUnit.MILLISECONDS); + return executor.scheduleWithFixedDelay(errorHandlingTask(task, true), initialDelay, delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); @@ -196,11 +213,38 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport public ScheduledFuture scheduleWithFixedDelay(Runnable task, long delay) { ScheduledExecutorService executor = getScheduledExecutor(); try { - return executor.scheduleWithFixedDelay(task, 0, delay, TimeUnit.MILLISECONDS); + return executor.scheduleWithFixedDelay(errorHandlingTask(task, true), 0, delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex); } } + private Runnable errorHandlingTask(Runnable task, boolean isRepeatingTask) { + return TaskUtils.errorHandlingTask(task, this.errorHandler, isRepeatingTask); + } + + + private static class DelegatingErrorHandlingCallable implements Callable { + + private final Callable delegate; + + private final ErrorHandler errorHandler; + + DelegatingErrorHandlingCallable(Callable delegate, ErrorHandler errorHandler) { + this.delegate = delegate; + this.errorHandler = errorHandler; + } + + public V call() throws Exception { + try { + return delegate.call(); + } + catch (Throwable t) { + this.errorHandler.handleError(t); + return null; + } + } + } + } diff --git a/org.springframework.context/src/main/java/org/springframework/scheduling/support/DelegatingErrorHandlingRunnable.java b/org.springframework.context/src/main/java/org/springframework/scheduling/support/DelegatingErrorHandlingRunnable.java new file mode 100644 index 00000000000..30da07db14e --- /dev/null +++ b/org.springframework.context/src/main/java/org/springframework/scheduling/support/DelegatingErrorHandlingRunnable.java @@ -0,0 +1,62 @@ +/* + * Copyright 2002-2009 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.scheduling.support; + +import org.springframework.util.Assert; + +/** + * Runnable wrapper that catches any exception or error thrown from its + * delegate Runnable and allows an {@link ErrorHandler} to handle it. + * + * @author Juergen Hoeller + * @author Mark Fisher + * @since 3.0 + */ +public class DelegatingErrorHandlingRunnable implements Runnable { + + private final Runnable delegate; + + private final ErrorHandler errorHandler; + + + /** + * Create a new DelegatingErrorHandlingRunnable. + * @param delegate the Runnable implementation to delegate to + * @param errorHandler the ErrorHandler for handling any exceptions + */ + public DelegatingErrorHandlingRunnable(Runnable delegate, ErrorHandler errorHandler) { + Assert.notNull(delegate, "Delegate must not be null"); + Assert.notNull(errorHandler, "ErrorHandler must not be null"); + this.delegate = delegate; + this.errorHandler = errorHandler; + } + + public void run() { + try { + this.delegate.run(); + } + catch (Throwable ex) { + this.errorHandler.handleError(ex); + } + } + + @Override + public String toString() { + return "DelegatingErrorHandlingRunnable for " + this.delegate; + } + +} diff --git a/org.springframework.context/src/main/java/org/springframework/scheduling/support/DelegatingExceptionProofRunnable.java b/org.springframework.context/src/main/java/org/springframework/scheduling/support/DelegatingExceptionProofRunnable.java deleted file mode 100644 index 79ad1734748..00000000000 --- a/org.springframework.context/src/main/java/org/springframework/scheduling/support/DelegatingExceptionProofRunnable.java +++ /dev/null @@ -1,91 +0,0 @@ -/* - * Copyright 2002-2009 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.scheduling.support; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; - -/** - * Runnable wrapper that catches any exception or error thrown - * from its delegate Runnable. Used for continuing scheduled - * execution even after an exception thrown from a task's Runnable. - * - * @author Juergen Hoeller - * @since 2.0.5 - */ -public class DelegatingExceptionProofRunnable implements Runnable { - - private static final Log logger = LogFactory.getLog(DelegatingExceptionProofRunnable.class); - - private final Runnable delegate; - - private final boolean propagateException; - - - /** - * Create a new DelegatingExceptionProofRunnable that logs the exception - * but isn't propagating it (in order to continue scheduled execution). - * @param delegate the Runnable implementation to delegate to - */ - public DelegatingExceptionProofRunnable(Runnable delegate) { - Assert.notNull(delegate, "Delegate must not be null"); - this.delegate = delegate; - this.propagateException = false; - } - - /** - * Create a new DelegatingExceptionProofRunnable. - * @param delegate the Runnable implementation to delegate to - * @param propagateException whether to propagate the exception after logging - * (note: this will typically cancel scheduled execution of the runnable) - */ - public DelegatingExceptionProofRunnable(Runnable delegate, boolean propagateException) { - Assert.notNull(delegate, "Delegate must not be null"); - this.delegate = delegate; - this.propagateException = propagateException; - } - - /** - * Return the wrapped Runnable implementation. - */ - public final Runnable getDelegate() { - return this.delegate; - } - - - public void run() { - try { - this.delegate.run(); - } - catch (Throwable ex) { - logger.error("Unexpected exception thrown from Runnable: " + this.delegate, ex); - if (this.propagateException) { - ReflectionUtils.rethrowRuntimeException(ex); - } - } - } - - - @Override - public String toString() { - return "DelegatingExceptionProofRunnable for " + this.delegate; - } - -} diff --git a/org.springframework.context/src/main/java/org/springframework/scheduling/support/ErrorHandler.java b/org.springframework.context/src/main/java/org/springframework/scheduling/support/ErrorHandler.java new file mode 100644 index 00000000000..783026afc80 --- /dev/null +++ b/org.springframework.context/src/main/java/org/springframework/scheduling/support/ErrorHandler.java @@ -0,0 +1,45 @@ +/* + * Copyright 2002-2009 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.scheduling.support; + +/** + * A strategy for handling errors that occur during asynchronous + * execution of tasks that have been submitted to a TaskScheduler. + * + * @author Mark Fisher + * @since 3.0. + */ +public interface ErrorHandler { + + /** + * An ErrorHandler strategy that will log the Exception but perform + * no further handling. This will suppress the error so that + * subsequent executions of the task will not be prevented. + */ + static final ErrorHandler LOG_AND_SUPPRESS = new LoggingErrorHandler(); + + /** + * An ErrorHandler strategy that will log at error level and then + * re-throw the Exception. Note: this will typically prevent subsequent + * execution of a scheduled task. + */ + static final ErrorHandler LOG_AND_PROPAGATE = new PropagatingErrorHandler(); + + + void handleError(Throwable t); + +} diff --git a/org.springframework.context/src/main/java/org/springframework/scheduling/support/LoggingErrorHandler.java b/org.springframework.context/src/main/java/org/springframework/scheduling/support/LoggingErrorHandler.java new file mode 100644 index 00000000000..8afdb2b973e --- /dev/null +++ b/org.springframework.context/src/main/java/org/springframework/scheduling/support/LoggingErrorHandler.java @@ -0,0 +1,40 @@ +/* + * Copyright 2002-2009 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.scheduling.support; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +/** + * An {@link ErrorHandler} implementation that logs the Throwable at error + * level. It does not perform any additional error handling. This can be + * useful when suppression of errors is the intended behavior. + * + * @author Mark Fisher + * @since 3.0 + */ +class LoggingErrorHandler implements ErrorHandler { + + private final Log logger = LogFactory.getLog(LoggingErrorHandler.class); + + public void handleError(Throwable t) { + if (logger.isErrorEnabled()) { + logger.error("Unexpected error occurred in scheduled task.", t); + } + } + +} diff --git a/org.springframework.context/src/main/java/org/springframework/scheduling/support/PropagatingErrorHandler.java b/org.springframework.context/src/main/java/org/springframework/scheduling/support/PropagatingErrorHandler.java new file mode 100644 index 00000000000..c91ba940666 --- /dev/null +++ b/org.springframework.context/src/main/java/org/springframework/scheduling/support/PropagatingErrorHandler.java @@ -0,0 +1,35 @@ +/* + * Copyright 2002-2009 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.scheduling.support; + +import org.springframework.util.ReflectionUtils; + +/** + * An {@link ErrorHandler} implementation that logs the Throwable at error + * level and then propagates it. + * + * @author Mark Fisher + * @since 3.0 + */ +class PropagatingErrorHandler extends LoggingErrorHandler { + + public void handleError(Throwable t) { + super.handleError(t); + ReflectionUtils.rethrowRuntimeException(t); + } + +}