16216864267

Committed 11 Jul 2025 09:36AM UTC coverage: 81.773% (+0.001%) from 81.772%

Build # 16216864267

Build Type

push

github

Committed by

TAKETODAY

Commit Message

:arrow_up: 更新 Gradle 8.14.3

Coverage Stats

59419 of 77611 branches covered (76.56%)

Branch coverage included in aggregate %.

140693 of 167106 relevant lines covered (84.19%)

3.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.24

today-context/src/main/java/infra/scheduling/concurrent/ThreadPoolTaskScheduler.java

/*
 * Copyright 2017 - 2024 the original author or authors.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see [https://www.gnu.org/licenses/]
 */

package infra.scheduling.concurrent;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.Callable;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableScheduledFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import infra.core.task.AsyncTaskExecutor;
import infra.core.task.TaskDecorator;
import infra.core.task.TaskRejectedException;
import infra.lang.Assert;
import infra.lang.Nullable;
import infra.scheduling.SchedulingTaskExecutor;
import infra.scheduling.TaskScheduler;
import infra.scheduling.Trigger;
import infra.scheduling.support.TaskUtils;
import infra.util.ConcurrentReferenceHashMap;
import infra.util.ErrorHandler;
import infra.util.concurrent.Future;
import infra.util.concurrent.ListenableFutureTask;

/**
 * A standard implementation of Infra {@link TaskScheduler} interface, wrapping
 * a native {@link java.util.concurrent.ScheduledThreadPoolExecutor} and providing
 * all applicable configuration options for it. The default number of scheduler
 * threads is 1; a higher number can be configured through {@link #setPoolSize}.
 *
 * <p>This is Infra traditional scheduler variant, staying as close as possible to
 * {@link java.util.concurrent.ScheduledExecutorService} semantics. Task execution happens
 * on the scheduler thread(s) rather than on separate execution threads. As a consequence,
 * a {@link ScheduledFuture} handle (e.g. from {@link #schedule(Runnable, Instant)})
 * represents the actual completion of the provided task (or series of repeated tasks).
 *
 * @author Juergen Hoeller
 * @author Mark Fisher
 * @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
 * @see #setPoolSize
 * @see #setRemoveOnCancelPolicy
 * @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
 * @see #setExecuteExistingDelayedTasksAfterShutdownPolicy
 * @see #setThreadFactory
 * @see #setErrorHandler
 * @since 4.0
 */
@SuppressWarnings("serial")
public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport implements AsyncTaskExecutor, SchedulingTaskExecutor, TaskScheduler {

  private static final TimeUnit NANO = TimeUnit.NANOSECONDS;

  private volatile int poolSize = 1;

  private volatile boolean removeOnCancelPolicy;

  private volatile boolean continueExistingPeriodicTasksAfterShutdownPolicy;

  private volatile boolean executeExistingDelayedTasksAfterShutdownPolicy = true;

  @Nullable
  private TaskDecorator taskDecorator;

  @Nullable
  private volatile ErrorHandler errorHandler;

  private Clock clock = Clock.systemDefaultZone();

  @Nullable
  private ScheduledExecutorService scheduledExecutor;

  // Underlying ScheduledFutureTask to user-level ListenableFuture handle, if any
  private final ConcurrentReferenceHashMap<Object, Future<?>> listenableFutureMap =
          new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);

  /**
   * Set the ScheduledExecutorService's pool size.
   * Default is 1.
   * <p><b>This setting can be modified at runtime, for example through JMX.</b>
   */
  public void setPoolSize(int poolSize) {
    Assert.isTrue(poolSize > 0, "'poolSize' must be 1 or higher");
    if (scheduledExecutor instanceof ScheduledThreadPoolExecutor tpe) {
      tpe.setCorePoolSize(poolSize);
    }
    this.poolSize = poolSize;
  }

  /**
   * Set the remove-on-cancel mode on {@link ScheduledThreadPoolExecutor}.
   * <p>Default is {@code false}. If set to {@code true}, the target executor will be
   * switched into remove-on-cancel mode (if possible).
   * <p><b>This setting can be modified at runtime, for example through JMX.</b>
   *
   * @see ScheduledThreadPoolExecutor#setRemoveOnCancelPolicy
   */
  public void setRemoveOnCancelPolicy(boolean flag) {
    if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor stpe) {
      stpe.setRemoveOnCancelPolicy(flag);
    }
    this.removeOnCancelPolicy = flag;
  }

  /**
   * Set whether to continue existing periodic tasks even when this executor has been shutdown.
   * <p>Default is {@code false}. If set to {@code true}, the target executor will be
   * switched into continuing periodic tasks (if possible).
   * <p><b>This setting can be modified at runtime, for example through JMX.</b>
   *
   * @see ScheduledThreadPoolExecutor#setContinueExistingPeriodicTasksAfterShutdownPolicy
   */
  public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean flag) {
    if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor tpe) {
      tpe.setContinueExistingPeriodicTasksAfterShutdownPolicy(flag);
    }
    this.continueExistingPeriodicTasksAfterShutdownPolicy = flag;
  }

  /**
   * Set whether to execute existing delayed tasks even when this executor has been shutdown.
   * <p>Default is {@code true}. If set to {@code false}, the target executor will be
   * switched into dropping remaining tasks (if possible).
   * <p><b>This setting can be modified at runtime, for example through JMX.</b>
   *
   * @see ScheduledThreadPoolExecutor#setExecuteExistingDelayedTasksAfterShutdownPolicy
   */
  public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean flag) {
    if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor tpe) {
      tpe.setExecuteExistingDelayedTasksAfterShutdownPolicy(flag);
    }
    this.executeExistingDelayedTasksAfterShutdownPolicy = flag;
  }

  /**
   * Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
   * about to be executed.
   * <p>Note that such a decorator is not being applied to the user-supplied
   * {@code Runnable}/{@code Callable} but rather to the scheduled execution
   * callback (a wrapper around the user-supplied task).
   * <p>The primary use case is to set some execution context around the task's
   * invocation, or to provide some monitoring/statistics for task execution.
   */
  public void setTaskDecorator(TaskDecorator taskDecorator) {
    this.taskDecorator = taskDecorator;
  }

  /**
   * Set a custom {@link ErrorHandler} strategy.
   */
  public void setErrorHandler(ErrorHandler errorHandler) {
    this.errorHandler = errorHandler;
  }

  /**
   * Set the clock to use for scheduling purposes.
   * <p>The default clock is the system clock for the default time zone.
   *
   * @see Clock#systemDefaultZone()
   */
  public void setClock(Clock clock) {
    Assert.notNull(clock, "Clock is required");
    this.clock = clock;
  }

  @Override
  public Clock getClock() {
    return this.clock;
  }

  @Override
  protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {
    ScheduledExecutorService executor = createExecutor(this.poolSize, threadFactory, rejectedHandler);
    if (executor instanceof ScheduledThreadPoolExecutor tpExecutor) {
      if (this.removeOnCancelPolicy) {
        tpExecutor.setRemoveOnCancelPolicy(true);
      }
      if (this.continueExistingPeriodicTasksAfterShutdownPolicy) {
        tpExecutor.setContinueExistingPeriodicTasksAfterShutdownPolicy(true);
      }
      if (!this.executeExistingDelayedTasksAfterShutdownPolicy) {
        tpExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
      }
    }
    this.scheduledExecutor = executor;
    return executor;
  }

  /**
   * Create a new {@link ScheduledExecutorService} instance.
   * <p>The default implementation creates a {@link ScheduledThreadPoolExecutor}.
   * Can be overridden in subclasses to provide custom {@link ScheduledExecutorService} instances.
   *
   * @param poolSize the specified pool size
   * @param threadFactory the ThreadFactory to use
   * @param rejectedHandler the RejectedExecutionHandler to use
   * @return a new ScheduledExecutorService instance
   * @see #afterPropertiesSet()
   * @see java.util.concurrent.ScheduledThreadPoolExecutor
   */
  protected ScheduledExecutorService createExecutor(int poolSize,
          ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {

    return new ScheduledThreadPoolExecutor(poolSize, threadFactory, rejectedHandler) {
      @Override
      protected void beforeExecute(Thread thread, Runnable task) {
        ThreadPoolTaskScheduler.this.beforeExecute(thread, task);
      }

      @Override
      protected void afterExecute(Runnable task, Throwable ex) {
        ThreadPoolTaskScheduler.this.afterExecute(task, ex);
      }

      @Override
      protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
        return decorateTaskIfNecessary(task);
      }

      @Override
      protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
        return decorateTaskIfNecessary(task);
      }
    };
  }

  /**
   * Return the underlying ScheduledExecutorService for native access.
   *
   * @return the underlying ScheduledExecutorService (never {@code null})
   * @throws IllegalStateException if the ThreadPoolTaskScheduler hasn't been initialized yet
   */
  public ScheduledExecutorService getScheduledExecutor() throws IllegalStateException {
    Assert.state(scheduledExecutor != null, "ThreadPoolTaskScheduler not initialized");
    return scheduledExecutor;
  }

  /**
   * Return the underlying ScheduledThreadPoolExecutor, if available.
   *
   * @return the underlying ScheduledExecutorService (never {@code null})
   * @throws IllegalStateException if the ThreadPoolTaskScheduler hasn't been initialized yet
   * or if the underlying ScheduledExecutorService isn't a ScheduledThreadPoolExecutor
   * @see #getScheduledExecutor()
   */
  public ScheduledThreadPoolExecutor getScheduledThreadPoolExecutor() throws IllegalStateException {
    Assert.state(scheduledExecutor instanceof ScheduledThreadPoolExecutor,
            "No ScheduledThreadPoolExecutor available");
    return (ScheduledThreadPoolExecutor) scheduledExecutor;
  }

  /**
   * Return the current pool size.
   * <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
   *
   * @see #getScheduledThreadPoolExecutor()
   * @see java.util.concurrent.ScheduledThreadPoolExecutor#getPoolSize()
   */
  public int getPoolSize() {
    if (scheduledExecutor == null) {
      // Not initialized yet: assume initial pool size.
      return poolSize;
    }
    return getScheduledThreadPoolExecutor().getPoolSize();
  }

  /**
   * Return the number of currently active threads.
   * <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
   *
   * @see #getScheduledThreadPoolExecutor()
   * @see java.util.concurrent.ScheduledThreadPoolExecutor#getActiveCount()
   */
  public int getActiveCount() {
    if (scheduledExecutor == null) {
      // Not initialized yet: assume no active threads.
      return 0;
    }
    return getScheduledThreadPoolExecutor().getActiveCount();
  }

  /**
   * Return the current setting for the remove-on-cancel mode.
   * <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
   */
  public boolean isRemoveOnCancelPolicy() {
    if (scheduledExecutor == null) {
      // Not initialized yet: return our setting for the time being.
      return removeOnCancelPolicy;
    }
    return getScheduledThreadPoolExecutor().getRemoveOnCancelPolicy();
  }

  // SchedulingTaskExecutor implementation

  @Override
  public void execute(Runnable task) {
    Executor executor = getScheduledExecutor();
    try {
      executor.execute(errorHandlingTask(task, false));
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public Future<Void> submit(Runnable task) {
    ExecutorService executor = getScheduledExecutor();
    try {
      var future = Future.<Void>forFutureTask(errorHandlingTask(task, false), executor);
      executeAndTrack(executor, future);
      return future;
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public <T> Future<T> submit(Callable<T> task) {
    ExecutorService executor = getScheduledExecutor();
    try {
      var future = Future.forFutureTask(new DelegatingErrorHandlingCallable<>(task, this.errorHandler), executor);
      executeAndTrack(executor, future);
      return future;
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  private void executeAndTrack(ExecutorService executor, ListenableFutureTask<?> task) {
    var scheduledFuture = executor.submit(task);
    listenableFutureMap.put(scheduledFuture, task);
    task.onCompleted(f -> listenableFutureMap.remove(scheduledFuture));
  }

  @Override
  protected void cancelRemainingTask(Runnable task) {
    super.cancelRemainingTask(task);
    // Cancel associated user-level ListenableFuture handle as well
    Future<?> future = this.listenableFutureMap.get(task);
    if (future != null) {
      future.cancel(true);
    }
  }

  // TaskScheduler implementation

  @Override
  @Nullable
  public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
    ScheduledExecutorService executor = getScheduledExecutor();
    try {
      ErrorHandler errorHandler = this.errorHandler;
      if (errorHandler == null) {
        errorHandler = TaskUtils.getDefaultErrorHandler(true);
      }
      return new ReschedulingRunnable(task, trigger, this.clock, executor, errorHandler).schedule();
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> schedule(Runnable task, Instant startTime) {
    ScheduledExecutorService executor = getScheduledExecutor();
    Duration delay = Duration.between(this.clock.instant(), startTime);
    try {
      return executor.schedule(errorHandlingTask(task, false), NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) {
    ScheduledExecutorService executor = getScheduledExecutor();
    Duration initialDelay = Duration.between(this.clock.instant(), startTime);
    try {
      return executor.scheduleAtFixedRate(errorHandlingTask(task, true),
              NANO.convert(initialDelay), NANO.convert(period), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) {
    ScheduledExecutorService executor = getScheduledExecutor();
    try {
      return executor.scheduleAtFixedRate(errorHandlingTask(task, true),
              0, NANO.convert(period), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) {
    ScheduledExecutorService executor = getScheduledExecutor();
    Duration initialDelay = Duration.between(this.clock.instant(), startTime);
    try {
      return executor.scheduleWithFixedDelay(errorHandlingTask(task, true),
              NANO.convert(initialDelay), NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) {
    ScheduledExecutorService executor = getScheduledExecutor();
    try {
      return executor.scheduleWithFixedDelay(errorHandlingTask(task, true),
              0, NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(executor, task, ex);
    }
  }

  private <V> RunnableScheduledFuture<V> decorateTaskIfNecessary(RunnableScheduledFuture<V> future) {
    return taskDecorator != null ? new DelegatingRunnableScheduledFuture<>(future, taskDecorator) : future;
  }

  private Runnable errorHandlingTask(Runnable task, boolean isRepeatingTask) {
    return TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, isRepeatingTask);
  }

  private static class DelegatingRunnableScheduledFuture<V> implements RunnableScheduledFuture<V> {

    private final RunnableScheduledFuture<V> future;

    private final Runnable decoratedRunnable;

    public DelegatingRunnableScheduledFuture(RunnableScheduledFuture<V> future, TaskDecorator taskDecorator) {
      this.future = future;
      this.decoratedRunnable = taskDecorator.decorate(this.future);
    }

    @Override
    public void run() {
      this.decoratedRunnable.run();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
      return this.future.cancel(mayInterruptIfRunning);
    }

    @Override
    public boolean isCancelled() {
      return this.future.isCancelled();
    }

    @Override
    public boolean isDone() {
      return this.future.isDone();
    }

    @Override
    public V get() throws InterruptedException, ExecutionException {
      return this.future.get();
    }

    @Override
    public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
      return this.future.get(timeout, unit);
    }

    @Override
    public boolean isPeriodic() {
      return this.future.isPeriodic();
    }

    @Override
    public long getDelay(TimeUnit unit) {
      return this.future.getDelay(unit);
    }

    @Override
    public int compareTo(Delayed o) {
      return this.future.compareTo(o);
    }

  }

}

1	/*
2	* Copyright 2017 - 2024 the original author or authors.
3	*
4	* This program is free software: you can redistribute it and/or modify
5	* it under the terms of the GNU General Public License as published by
6	* the Free Software Foundation, either version 3 of the License, or
7	* (at your option) any later version.
8	*
9	* This program is distributed in the hope that it will be useful,
10	* but WITHOUT ANY WARRANTY; without even the implied warranty of
11	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	* GNU General Public License for more details.
13	*
14	* You should have received a copy of the GNU General Public License
15	* along with this program. If not, see [https://www.gnu.org/licenses/]
16	*/
17
18	package infra.scheduling.concurrent;
19
20	import java.time.Clock;
21	import java.time.Duration;
22	import java.time.Instant;
23	import java.util.concurrent.Callable;
24	import java.util.concurrent.Delayed;
25	import java.util.concurrent.ExecutionException;
26	import java.util.concurrent.Executor;
27	import java.util.concurrent.ExecutorService;
28	import java.util.concurrent.RejectedExecutionException;
29	import java.util.concurrent.RejectedExecutionHandler;
30	import java.util.concurrent.RunnableScheduledFuture;
31	import java.util.concurrent.ScheduledExecutorService;
32	import java.util.concurrent.ScheduledFuture;
33	import java.util.concurrent.ScheduledThreadPoolExecutor;
34	import java.util.concurrent.ThreadFactory;
35	import java.util.concurrent.TimeUnit;
36	import java.util.concurrent.TimeoutException;
37
38	import infra.core.task.AsyncTaskExecutor;
39	import infra.core.task.TaskDecorator;
40	import infra.core.task.TaskRejectedException;
41	import infra.lang.Assert;
42	import infra.lang.Nullable;
43	import infra.scheduling.SchedulingTaskExecutor;
44	import infra.scheduling.TaskScheduler;
45	import infra.scheduling.Trigger;
46	import infra.scheduling.support.TaskUtils;
47	import infra.util.ConcurrentReferenceHashMap;
48	import infra.util.ErrorHandler;
49	import infra.util.concurrent.Future;
50	import infra.util.concurrent.ListenableFutureTask;
51
52	/**
53	* A standard implementation of Infra {@link TaskScheduler} interface, wrapping
54	* a native {@link java.util.concurrent.ScheduledThreadPoolExecutor} and providing
55	* all applicable configuration options for it. The default number of scheduler
56	* threads is 1; a higher number can be configured through {@link #setPoolSize}.
57	*
58	* <p>This is Infra traditional scheduler variant, staying as close as possible to
59	* {@link java.util.concurrent.ScheduledExecutorService} semantics. Task execution happens
60	* on the scheduler thread(s) rather than on separate execution threads. As a consequence,
61	* a {@link ScheduledFuture} handle (e.g. from {@link #schedule(Runnable, Instant)})
62	* represents the actual completion of the provided task (or series of repeated tasks).
63	*
64	* @author Juergen Hoeller
65	* @author Mark Fisher
66	* @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
67	* @see #setPoolSize
68	* @see #setRemoveOnCancelPolicy
69	* @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
70	* @see #setExecuteExistingDelayedTasksAfterShutdownPolicy
71	* @see #setThreadFactory
72	* @see #setErrorHandler
73	* @since 4.0
74	*/
75	@SuppressWarnings("serial")
76	public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport implements AsyncTaskExecutor, SchedulingTaskExecutor, TaskScheduler {	2✔
77
78	private static final TimeUnit NANO = TimeUnit.NANOSECONDS;	3✔
79
80	private volatile int poolSize = 1;	3✔
81
82	private volatile boolean removeOnCancelPolicy;
83
84	private volatile boolean continueExistingPeriodicTasksAfterShutdownPolicy;
85
86	private volatile boolean executeExistingDelayedTasksAfterShutdownPolicy = true;	3✔
87
88	@Nullable
89	private TaskDecorator taskDecorator;
90
91	@Nullable
92	private volatile ErrorHandler errorHandler;
93
94	private Clock clock = Clock.systemDefaultZone();	3✔
95
96	@Nullable
97	private ScheduledExecutorService scheduledExecutor;
98
99	// Underlying ScheduledFutureTask to user-level ListenableFuture handle, if any
100	private final ConcurrentReferenceHashMap<Object, Future<?>> listenableFutureMap =	8✔
101	new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
102
103	/**
104	* Set the ScheduledExecutorService's pool size.
105	* Default is 1.
106	* <p><b>This setting can be modified at runtime, for example through JMX.</b>
107	*/
108	public void setPoolSize(int poolSize) {
109	Assert.isTrue(poolSize > 0, "'poolSize' must be 1 or higher");	6!
110	if (scheduledExecutor instanceof ScheduledThreadPoolExecutor tpe) {	6!
111	tpe.setCorePoolSize(poolSize);	×
112	}
113	this.poolSize = poolSize;	3✔
114	}	1✔
115
116	/**
117	* Set the remove-on-cancel mode on {@link ScheduledThreadPoolExecutor}.
118	* <p>Default is {@code false}. If set to {@code true}, the target executor will be
119	* switched into remove-on-cancel mode (if possible).
120	* <p><b>This setting can be modified at runtime, for example through JMX.</b>
121	*
122	* @see ScheduledThreadPoolExecutor#setRemoveOnCancelPolicy
123	*/
124	public void setRemoveOnCancelPolicy(boolean flag) {
125	if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor stpe) {	×
126	stpe.setRemoveOnCancelPolicy(flag);	×
127	}
128	this.removeOnCancelPolicy = flag;	×
129	}	×
130
131	/**
132	* Set whether to continue existing periodic tasks even when this executor has been shutdown.
133	* <p>Default is {@code false}. If set to {@code true}, the target executor will be
134	* switched into continuing periodic tasks (if possible).
135	* <p><b>This setting can be modified at runtime, for example through JMX.</b>
136	*
137	* @see ScheduledThreadPoolExecutor#setContinueExistingPeriodicTasksAfterShutdownPolicy
138	*/
139	public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean flag) {
140	if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor tpe) {	×
141	tpe.setContinueExistingPeriodicTasksAfterShutdownPolicy(flag);	×
142	}
143	this.continueExistingPeriodicTasksAfterShutdownPolicy = flag;	×
144	}	×
145
146	/**
147	* Set whether to execute existing delayed tasks even when this executor has been shutdown.
148	* <p>Default is {@code true}. If set to {@code false}, the target executor will be
149	* switched into dropping remaining tasks (if possible).
150	* <p><b>This setting can be modified at runtime, for example through JMX.</b>
151	*
152	* @see ScheduledThreadPoolExecutor#setExecuteExistingDelayedTasksAfterShutdownPolicy
153	*/
154	public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean flag) {
155	if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor tpe) {	×
156	tpe.setExecuteExistingDelayedTasksAfterShutdownPolicy(flag);	×
157	}
158	this.executeExistingDelayedTasksAfterShutdownPolicy = flag;	×
159	}	×
160
161	/**
162	* Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
163	* about to be executed.
164	* <p>Note that such a decorator is not being applied to the user-supplied
165	* {@code Runnable}/{@code Callable} but rather to the scheduled execution
166	* callback (a wrapper around the user-supplied task).
167	* <p>The primary use case is to set some execution context around the task's
168	* invocation, or to provide some monitoring/statistics for task execution.
169	*/
170	public void setTaskDecorator(TaskDecorator taskDecorator) {
171	this.taskDecorator = taskDecorator;	3✔
172	}	1✔
173
174	/**
175	* Set a custom {@link ErrorHandler} strategy.
176	*/
177	public void setErrorHandler(ErrorHandler errorHandler) {
178	this.errorHandler = errorHandler;	3✔
179	}	1✔
180
181	/**
182	* Set the clock to use for scheduling purposes.
183	* <p>The default clock is the system clock for the default time zone.
184	*
185	* @see Clock#systemDefaultZone()
186	*/
187	public void setClock(Clock clock) {
188	Assert.notNull(clock, "Clock is required");	×
189	this.clock = clock;	×
190	}	×
191
192	@Override
193	public Clock getClock() {
194	return this.clock;	3✔
195	}
196
197	@Override
198	protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {
199	ScheduledExecutorService executor = createExecutor(this.poolSize, threadFactory, rejectedHandler);	7✔
200	if (executor instanceof ScheduledThreadPoolExecutor tpExecutor) {	6!
201	if (this.removeOnCancelPolicy) {	3!
202	tpExecutor.setRemoveOnCancelPolicy(true);	×
203	}
204	if (this.continueExistingPeriodicTasksAfterShutdownPolicy) {	3!
205	tpExecutor.setContinueExistingPeriodicTasksAfterShutdownPolicy(true);	×
206	}
207	if (!this.executeExistingDelayedTasksAfterShutdownPolicy) {	3!
208	tpExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);	×
209	}
210	}
211	this.scheduledExecutor = executor;	3✔
212	return executor;	2✔
213	}
214
215	/**
216	* Create a new {@link ScheduledExecutorService} instance.
217	* <p>The default implementation creates a {@link ScheduledThreadPoolExecutor}.
218	* Can be overridden in subclasses to provide custom {@link ScheduledExecutorService} instances.
219	*
220	* @param poolSize the specified pool size
221	* @param threadFactory the ThreadFactory to use
222	* @param rejectedHandler the RejectedExecutionHandler to use
223	* @return a new ScheduledExecutorService instance
224	* @see #afterPropertiesSet()
225	* @see java.util.concurrent.ScheduledThreadPoolExecutor
226	*/
227	protected ScheduledExecutorService createExecutor(int poolSize,
228	ThreadFactory threadFactory, RejectedExecutionHandler rejectedHandler) {
229
230	return new ScheduledThreadPoolExecutor(poolSize, threadFactory, rejectedHandler) {	17✔
231	@Override
232	protected void beforeExecute(Thread thread, Runnable task) {
233	ThreadPoolTaskScheduler.this.beforeExecute(thread, task);	5✔
234	}	1✔
235
236	@Override
237	protected void afterExecute(Runnable task, Throwable ex) {
238	ThreadPoolTaskScheduler.this.afterExecute(task, ex);	5✔
239	}	1✔
240
241	@Override
242	protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
243	return decorateTaskIfNecessary(task);	5✔
244	}
245
246	@Override
247	protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task) {
248	return decorateTaskIfNecessary(task);	×
249	}
250	};
251	}
252
253	/**
254	* Return the underlying ScheduledExecutorService for native access.
255	*
256	* @return the underlying ScheduledExecutorService (never {@code null})
257	* @throws IllegalStateException if the ThreadPoolTaskScheduler hasn't been initialized yet
258	*/
259	public ScheduledExecutorService getScheduledExecutor() throws IllegalStateException {
260	Assert.state(scheduledExecutor != null, "ThreadPoolTaskScheduler not initialized");	7!
261	return scheduledExecutor;	3✔
262	}
263
264	/**
265	* Return the underlying ScheduledThreadPoolExecutor, if available.
266	*
267	* @return the underlying ScheduledExecutorService (never {@code null})
268	* @throws IllegalStateException if the ThreadPoolTaskScheduler hasn't been initialized yet
269	* or if the underlying ScheduledExecutorService isn't a ScheduledThreadPoolExecutor
270	* @see #getScheduledExecutor()
271	*/
272	public ScheduledThreadPoolExecutor getScheduledThreadPoolExecutor() throws IllegalStateException {
273	Assert.state(scheduledExecutor instanceof ScheduledThreadPoolExecutor,	×
274	"No ScheduledThreadPoolExecutor available");
275	return (ScheduledThreadPoolExecutor) scheduledExecutor;	×
276	}
277
278	/**
279	* Return the current pool size.
280	* <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
281	*
282	* @see #getScheduledThreadPoolExecutor()
283	* @see java.util.concurrent.ScheduledThreadPoolExecutor#getPoolSize()
284	*/
285	public int getPoolSize() {
286	if (scheduledExecutor == null) {	3!
287	// Not initialized yet: assume initial pool size.
288	return poolSize;	3✔
289	}
290	return getScheduledThreadPoolExecutor().getPoolSize();	×
291	}
292
293	/**
294	* Return the number of currently active threads.
295	* <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
296	*
297	* @see #getScheduledThreadPoolExecutor()
298	* @see java.util.concurrent.ScheduledThreadPoolExecutor#getActiveCount()
299	*/
300	public int getActiveCount() {
301	if (scheduledExecutor == null) {	×
302	// Not initialized yet: assume no active threads.
303	return 0;	×
304	}
305	return getScheduledThreadPoolExecutor().getActiveCount();	×
306	}
307
308	/**
309	* Return the current setting for the remove-on-cancel mode.
310	* <p>Requires an underlying {@link ScheduledThreadPoolExecutor}.
311	*/
312	public boolean isRemoveOnCancelPolicy() {
313	if (scheduledExecutor == null) {	×
314	// Not initialized yet: return our setting for the time being.
315	return removeOnCancelPolicy;	×
316	}
317	return getScheduledThreadPoolExecutor().getRemoveOnCancelPolicy();	×
318	}
319
320	// SchedulingTaskExecutor implementation
321
322	@Override
323	public void execute(Runnable task) {
324	Executor executor = getScheduledExecutor();	3✔
325	try {
326	executor.execute(errorHandlingTask(task, false));	6✔
327	}
328	catch (RejectedExecutionException ex) {	×
329	throw new TaskRejectedException(executor, task, ex);	×
330	}	1✔
331	}	1✔
332
333	@Override
334	public Future<Void> submit(Runnable task) {
335	ExecutorService executor = getScheduledExecutor();	3✔
336	try {
337	var future = Future.<Void>forFutureTask(errorHandlingTask(task, false), executor);	7✔
338	executeAndTrack(executor, future);	4✔
339	return future;	2✔
340	}
341	catch (RejectedExecutionException ex) {	×
342	throw new TaskRejectedException(executor, task, ex);	×
343	}
344	}
345
346	@Override
347	public <T> Future<T> submit(Callable<T> task) {
348	ExecutorService executor = getScheduledExecutor();	3✔
349	try {
350	var future = Future.forFutureTask(new DelegatingErrorHandlingCallable<>(task, this.errorHandler), executor);	9✔
351	executeAndTrack(executor, future);	4✔
352	return future;	2✔
353	}
354	catch (RejectedExecutionException ex) {	×
355	throw new TaskRejectedException(executor, task, ex);	×
356	}
357	}
358
359	private void executeAndTrack(ExecutorService executor, ListenableFutureTask<?> task) {
360	var scheduledFuture = executor.submit(task);	4✔
361	listenableFutureMap.put(scheduledFuture, task);	6✔
362	task.onCompleted(f -> listenableFutureMap.remove(scheduledFuture));	12✔
363	}	1✔
364
365	@Override
366	protected void cancelRemainingTask(Runnable task) {
367	super.cancelRemainingTask(task);	3✔
368	// Cancel associated user-level ListenableFuture handle as well
369	Future<?> future = this.listenableFutureMap.get(task);	6✔
370	if (future != null) {	2✔
371	future.cancel(true);	4✔
372	}
373	}	1✔
374
375	// TaskScheduler implementation
376
377	@Override
378	@Nullable
379	public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
380	ScheduledExecutorService executor = getScheduledExecutor();	3✔
381	try {
382	ErrorHandler errorHandler = this.errorHandler;	3✔
383	if (errorHandler == null) {	2!
384	errorHandler = TaskUtils.getDefaultErrorHandler(true);	3✔
385	}
386	return new ReschedulingRunnable(task, trigger, this.clock, executor, errorHandler).schedule();	11✔
387	}
388	catch (RejectedExecutionException ex) {	×
389	throw new TaskRejectedException(executor, task, ex);	×
390	}
391	}
392
393	@Override
394	public ScheduledFuture<?> schedule(Runnable task, Instant startTime) {
395	ScheduledExecutorService executor = getScheduledExecutor();	3✔
396	Duration delay = Duration.between(this.clock.instant(), startTime);	6✔
397	try {
398	return executor.schedule(errorHandlingTask(task, false), NANO.convert(delay), NANO);	11✔
399	}
400	catch (RejectedExecutionException ex) {	×
401	throw new TaskRejectedException(executor, task, ex);	×
402	}
403	}
404
405	@Override
406	public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) {
407	ScheduledExecutorService executor = getScheduledExecutor();	3✔
408	Duration initialDelay = Duration.between(this.clock.instant(), startTime);	6✔
409	try {
410	return executor.scheduleAtFixedRate(errorHandlingTask(task, true),	9✔
411	NANO.convert(initialDelay), NANO.convert(period), NANO);	5✔
412	}
413	catch (RejectedExecutionException ex) {	×
414	throw new TaskRejectedException(executor, task, ex);	×
415	}
416	}
417
418	@Override
419	public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) {
420	ScheduledExecutorService executor = getScheduledExecutor();	3✔
421	try {
422	return executor.scheduleAtFixedRate(errorHandlingTask(task, true),	10✔
423	0, NANO.convert(period), NANO);	2✔
424	}
425	catch (RejectedExecutionException ex) {	×
426	throw new TaskRejectedException(executor, task, ex);	×
427	}
428	}
429
430	@Override
431	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) {
432	ScheduledExecutorService executor = getScheduledExecutor();	3✔
433	Duration initialDelay = Duration.between(this.clock.instant(), startTime);	6✔
434	try {
435	return executor.scheduleWithFixedDelay(errorHandlingTask(task, true),	9✔
436	NANO.convert(initialDelay), NANO.convert(delay), NANO);	5✔
437	}
438	catch (RejectedExecutionException ex) {	×
439	throw new TaskRejectedException(executor, task, ex);	×
440	}
441	}
442
443	@Override
444	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) {
445	ScheduledExecutorService executor = getScheduledExecutor();	×
446	try {
447	return executor.scheduleWithFixedDelay(errorHandlingTask(task, true),	×
448	0, NANO.convert(delay), NANO);	×
449	}
450	catch (RejectedExecutionException ex) {	×
451	throw new TaskRejectedException(executor, task, ex);	×
452	}
453	}
454
455	private <V> RunnableScheduledFuture<V> decorateTaskIfNecessary(RunnableScheduledFuture<V> future) {
456	return taskDecorator != null ? new DelegatingRunnableScheduledFuture<>(future, taskDecorator) : future;	12✔
457	}
458
459	private Runnable errorHandlingTask(Runnable task, boolean isRepeatingTask) {
460	return TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, isRepeatingTask);	6✔
461	}
462
463	private static class DelegatingRunnableScheduledFuture<V> implements RunnableScheduledFuture<V> {
464
465	private final RunnableScheduledFuture<V> future;
466
467	private final Runnable decoratedRunnable;
468
469	public DelegatingRunnableScheduledFuture(RunnableScheduledFuture<V> future, TaskDecorator taskDecorator) {	2✔
470	this.future = future;	3✔
471	this.decoratedRunnable = taskDecorator.decorate(this.future);	6✔
472	}	1✔
473
474	@Override
475	public void run() {
476	this.decoratedRunnable.run();	3✔
477	}	1✔
478
479	@Override
480	public boolean cancel(boolean mayInterruptIfRunning) {
481	return this.future.cancel(mayInterruptIfRunning);	5✔
482	}
483
484	@Override
485	public boolean isCancelled() {
486	return this.future.isCancelled();	4✔
487	}
488
489	@Override
490	public boolean isDone() {
491	return this.future.isDone();	4✔
492	}
493
494	@Override
495	public V get() throws InterruptedException, ExecutionException {
496	return this.future.get();	×
497	}
498
499	@Override
500	public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
501	return this.future.get(timeout, unit);	6✔
502	}
503
504	@Override
505	public boolean isPeriodic() {
506	return this.future.isPeriodic();	×
507	}
508
509	@Override
510	public long getDelay(TimeUnit unit) {
511	return this.future.getDelay(unit);	5✔
512	}
513
514	@Override
515	public int compareTo(Delayed o) {
516	return this.future.compareTo(o);	5✔
517	}
518
519	}
520
521	}

TAKETODAY / today-infrastructure / 16216864267

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous