This class uses the {@link ScheduledExecutorService} returned from {@link #executor} to run * the {@link #startUp} and {@link #shutDown} methods and also uses that service to schedule the * {@link #runOneIteration} that will be executed periodically as specified by its * {@link Scheduler}. When this service is asked to stop via {@link #stop} or {@link #stopAndWait}, * it will cancel the periodic task (but not interrupt it) and wait for it to stop before running * the {@link #shutDown} method. * *
Subclasses are guaranteed that the life cycle methods ({@link #runOneIteration}, {@link * #startUp} and {@link #shutDown}) will never run concurrently. Notably, if any execution of {@link * #runOneIteration} takes longer than its schedule defines, then subsequent executions may start * late. Also, all life cycle methods are executed with a lock held, so subclasses can safely * modify shared state without additional synchronization necessary for visibility to later * executions of the life cycle methods. * *
{@code
* class CrawlingService extends AbstractScheduledService {
* private Set visited;
* private Queue toCrawl;
* protected void startUp() throws Exception {
* toCrawl = readStartingUris();
* }
*
* protected void runOneIteration() throws Exception {
* Uri uri = toCrawl.remove();
* Collection newUris = crawl(uri);
* visited.add(uri);
* for (Uri newUri : newUris) {
* if (!visited.contains(newUri)) { toCrawl.add(newUri); }
* }
* }
*
* protected void shutDown() throws Exception {
* saveUris(toCrawl);
* }
*
* protected Scheduler scheduler() {
* return Scheduler.newFixedRateSchedule(0, 1, TimeUnit.SECONDS);
* }
* }}
*
* This class uses the life cycle methods to read in a list of starting URIs and save the set of
* outstanding URIs when shutting down. Also, it takes advantage of the scheduling functionality to
* rate limit the number of queries we perform.
*
* @author Luke Sandberg
* @since 11.0
*/
@Beta
public abstract class AbstractScheduledService implements Service {
private static final Logger logger = Logger.getLogger(AbstractScheduledService.class.getName());
/**
* A scheduler defines the policy for how the {@link AbstractScheduledService} should run its
* task.
*
* Consider using the {@link #newFixedDelaySchedule} and {@link #newFixedRateSchedule} factory * methods, these provide {@link Scheduler} instances for the common use case of running the * service with a fixed schedule. If more flexibility is needed then consider subclassing the * {@link CustomScheduler} abstract class in preference to creating your own {@link Scheduler} * implementation. * * @author Luke Sandberg * @since 11.0 */ public abstract static class Scheduler { /** * Returns a {@link Scheduler} that schedules the task using the * {@link ScheduledExecutorService#scheduleWithFixedDelay} method. * * @param initialDelay the time to delay first execution * @param delay the delay between the termination of one execution and the commencement of the * next * @param unit the time unit of the initialDelay and delay parameters */ public static Scheduler newFixedDelaySchedule(final long initialDelay, final long delay, final TimeUnit unit) { return new Scheduler() { @Override public Future schedule(AbstractService service, ScheduledExecutorService executor, Runnable task) { return executor.scheduleWithFixedDelay(task, initialDelay, delay, unit); } }; } /** * Returns a {@link Scheduler} that schedules the task using the * {@link ScheduledExecutorService#scheduleAtFixedRate} method. * * @param initialDelay the time to delay first execution * @param period the period between successive executions of the task * @param unit the time unit of the initialDelay and period parameters */ public static Scheduler newFixedRateSchedule(final long initialDelay, final long period, final TimeUnit unit) { return new Scheduler() { @Override public Future schedule(AbstractService service, ScheduledExecutorService executor, Runnable task) { return executor.scheduleAtFixedRate(task, initialDelay, period, unit); } }; } /** Schedules the task to run on the provided executor on behalf of the service. */ abstract Future schedule(AbstractService service, ScheduledExecutorService executor, Runnable runnable); private Scheduler() {} } /* use AbstractService for state management */ private final AbstractService delegate = new AbstractService() { // A handle to the running task so that we can stop it when a shutdown has been requested. // These two fields are volatile because their values will be accessed from multiple threads. private volatile Future runningTask; private volatile ScheduledExecutorService executorService; // This lock protects the task so we can ensure that none of the template methods (startUp, // shutDown or runOneIteration) run concurrently with one another. private final ReentrantLock lock = new ReentrantLock(); private final Runnable task = new Runnable() { @Override public void run() { lock.lock(); try { AbstractScheduledService.this.runOneIteration(); } catch (Throwable t) { try { shutDown(); } catch (Exception ignored) { logger.log(Level.WARNING, "Error while attempting to shut down the service after failure.", ignored); } notifyFailed(t); throw Throwables.propagate(t); } finally { lock.unlock(); } } }; @Override protected final void doStart() { executorService = executor(); executorService.execute(new Runnable() { @Override public void run() { lock.lock(); try { startUp(); runningTask = scheduler().schedule(delegate, executorService, task); notifyStarted(); } catch (Throwable t) { notifyFailed(t); throw Throwables.propagate(t); } finally { lock.unlock(); } } }); } @Override protected final void doStop() { runningTask.cancel(false); executorService.execute(new Runnable() { @Override public void run() { try { lock.lock(); try { if (state() != State.STOPPING) { // This means that the state has changed since we were scheduled. This implies that // an execution of runOneIteration has thrown an exception and we have transitioned // to a failed state, also this means that shutDown has already been called, so we // do not want to call it again. return; } shutDown(); } finally { lock.unlock(); } notifyStopped(); } catch (Throwable t) { notifyFailed(t); throw Throwables.propagate(t); } } }); } }; /** * Run one iteration of the scheduled task. If any invocation of this method throws an exception, * the service will transition to the {@link Service.State#FAILED} state and this method will no * longer be called. */ protected abstract void runOneIteration() throws Exception; /** Start the service. */ protected abstract void startUp() throws Exception; /** Stop the service. This is guaranteed not to run concurrently with {@link #runOneIteration}. */ protected abstract void shutDown() throws Exception; /** * Returns the {@link Scheduler} object used to configure this service. This method will only be * called once. */ protected abstract Scheduler scheduler(); /** * Returns the {@link ScheduledExecutorService} that will be used to execute the {@link #startUp}, * {@link #runOneIteration} and {@link #shutDown} methods. The executor will not be * {@link ScheduledExecutorService#shutdown} when this service stops. Subclasses may override this * method to use a custom {@link ScheduledExecutorService} instance. * *
By default this returns a new {@link ScheduledExecutorService} with a single thread thread
* pool. This method will only be called once.
*/
protected ScheduledExecutorService executor() {
return Executors.newSingleThreadScheduledExecutor();
}
@Override public String toString() {
return getClass().getSimpleName() + " [" + state() + "]";
}
// We override instead of using ForwardingService so that these can be final.
@Override public final ListenableFuture This is guaranteed to be called immediately after the task has completed an iteration and
* on the same thread as the previous execution of {@link
* AbstractScheduledService#runOneIteration}.
*
* @return a schedule that defines the delay before the next execution.
*/
protected abstract Schedule getNextSchedule() throws Exception;
}
}