diff --git a/spring-context/src/main/java/org/springframework/scheduling/concurrent/ForkJoinPoolFactoryBean.java b/spring-context/src/main/java/org/springframework/scheduling/concurrent/ForkJoinPoolFactoryBean.java index fe30fd108be..d74f1e1f9af 100644 --- a/spring-context/src/main/java/org/springframework/scheduling/concurrent/ForkJoinPoolFactoryBean.java +++ b/spring-context/src/main/java/org/springframework/scheduling/concurrent/ForkJoinPoolFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -17,6 +17,7 @@ package org.springframework.scheduling.concurrent; import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.TimeUnit; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.FactoryBean; @@ -24,7 +25,7 @@ import org.springframework.beans.factory.InitializingBean; /** * A Spring {@link FactoryBean} that builds and exposes a preconfigured {@link ForkJoinPool}. - * May be used on Java 7 as well as on Java 6 with {@code jsr166.jar} on the classpath + * May be used on Java 7 and 8 as well as on Java 6 with {@code jsr166.jar} on the classpath * (ideally on the VM bootstrap classpath). * *

For details on the ForkJoinPool API and its use with RecursiveActions, see the @@ -38,6 +39,8 @@ import org.springframework.beans.factory.InitializingBean; */ public class ForkJoinPoolFactoryBean implements FactoryBean, InitializingBean, DisposableBean { + private boolean commonPool = false; + private int parallelism = Runtime.getRuntime().availableProcessors(); private ForkJoinPool.ForkJoinWorkerThreadFactory threadFactory = ForkJoinPool.defaultForkJoinWorkerThreadFactory; @@ -46,9 +49,28 @@ public class ForkJoinPoolFactoryBean implements FactoryBean, Initi private boolean asyncMode = false; + private int awaitTerminationSeconds = 0; + private ForkJoinPool forkJoinPool; + /** + * Set whether to expose JDK 8's 'common' {@link ForkJoinPool}. + *

Default is "false", creating a local {@link ForkJoinPool} instance based on the + * {@link #setParallelism "parallelism"}, {@link #setThreadFactory "threadFactory"}, + * {@link #setUncaughtExceptionHandler "uncaughtExceptionHandler"} and + * {@link #setAsyncMode "asyncMode"} properties on this FactoryBean. + *

NOTE: Setting this flag to "true" effectively ignores all other + * properties on this FactoryBean, reusing the shared common JDK {@link ForkJoinPool} + * instead. This is a fine choice on JDK 8 but does remove the application's ability + * to customize ForkJoinPool behavior, in particular the use of custom threads. + * @since 3.2 + * @see java.util.concurrent.ForkJoinPool#commonPool() + */ + public void setCommonPool(boolean commonPool) { + this.commonPool = commonPool; + } + /** * Specify the parallelism level. Default is {@link Runtime#availableProcessors()}. */ @@ -82,9 +104,30 @@ public class ForkJoinPoolFactoryBean implements FactoryBean, Initi this.asyncMode = asyncMode; } + /** + * Set the maximum number of seconds that this ForkJoinPool is supposed to block + * on shutdown in order to wait for remaining tasks to complete their execution + * before the rest of the container continues to shut down. This is particularly + * useful if your remaining tasks are likely to need access to other resources + * that are also managed by the container. + *

By default, this ForkJoinPool won't wait for the termination of tasks at all. + * It will continue to fully execute all ongoing tasks as well as all remaining + * tasks in the queue, in parallel to the rest of the container shutting down. + * In contrast, if you specify an await-termination period using this property, + * this executor will wait for the given time (max) for the termination of tasks. + *

Note that this feature works for the {@link #setCommonPool "commonPool"} + * mode as well. The underlying ForkJoinPool won't actually terminate in that + * case but will wait for all tasks to terminate. + * @see java.util.concurrent.ForkJoinPool#shutdown() + * @see java.util.concurrent.ForkJoinPool#awaitTermination + */ + public void setAwaitTerminationSeconds(int awaitTerminationSeconds) { + this.awaitTerminationSeconds = awaitTerminationSeconds; + } + public void afterPropertiesSet() { - this.forkJoinPool = - new ForkJoinPool(this.parallelism, this.threadFactory, this.uncaughtExceptionHandler, this.asyncMode); + this.forkJoinPool = (this.commonPool ? ForkJoinPool.commonPool() : + new ForkJoinPool(this.parallelism, this.threadFactory, this.uncaughtExceptionHandler, this.asyncMode)); } @@ -102,7 +145,18 @@ public class ForkJoinPoolFactoryBean implements FactoryBean, Initi public void destroy() { + // Ignored for the common pool. this.forkJoinPool.shutdown(); + + // Wait for all tasks to terminate - works for the common pool as well. + if (this.awaitTerminationSeconds > 0) { + try { + this.forkJoinPool.awaitTermination(this.awaitTerminationSeconds, TimeUnit.SECONDS); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } } }