/* * Copyright (C) 2011 The Guava 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 com.google.common.util.concurrent; import static com.google.common.util.concurrent.MoreExecutors.directExecutor; import com.google.common.annotations.Beta; import com.google.common.base.Preconditions; import com.google.common.base.Supplier; import com.google.common.base.Throwables; import java.util.concurrent.Callable; import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.locks.ReentrantLock; import java.util.logging.Level; import java.util.logging.Logger; import javax.annotation.concurrent.GuardedBy; /** * Base class for services that can implement {@link #startUp} and {@link #shutDown} but while in * the "running" state need to perform a periodic task. Subclasses can implement {@link #startUp}, * {@link #shutDown} and also a {@link #runOneIteration} method that will be executed periodically. * *
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 #stopAsync} 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. * *
Here is a sketch of a service which crawls a website and uses the scheduling capabilities to * rate limit itself.
{@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
* {@link CustomScheduler}.
*
* @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 = MoreExecutors.renamingDecorator(executor(), new Supplier By default this method does nothing.
*/
protected void startUp() throws Exception {}
/**
* Stop the service. This is guaranteed not to run concurrently with {@link #runOneIteration}.
*
* By default this method does nothing.
*/
protected 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. If this method is overridden the
* executor will not be {@linkplain ScheduledExecutorService#shutdown shutdown} when this
* service {@linkplain Service.State#TERMINATED terminates} or
* {@linkplain Service.State#TERMINATED fails}. Subclasses may override this method to supply a
* custom {@link ScheduledExecutorService} instance. This method is guaranteed to only be called
* once.
*
* By default this returns a new {@link ScheduledExecutorService} with a single thread thread
* pool that sets the name of the thread to the {@linkplain #serviceName() service name}.
* Also, the pool will be {@linkplain ScheduledExecutorService#shutdown() shut down} when the
* service {@linkplain Service.State#TERMINATED terminates} or
* {@linkplain Service.State#TERMINATED fails}.
*/
protected ScheduledExecutorService executor() {
final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
new ThreadFactory() {
@Override public Thread newThread(Runnable runnable) {
return MoreExecutors.newThread(serviceName(), runnable);
}
});
// Add a listener to shutdown the executor after the service is stopped. This ensures that the
// JVM shutdown will not be prevented from exiting after this service has stopped or failed.
// Technically this listener is added after start() was called so it is a little gross, but it
// is called within doStart() so we know that the service cannot terminate or fail concurrently
// with adding this listener so it is impossible to miss an event that we are interested in.
addListener(new Listener() {
@Override public void terminated(State from) {
executor.shutdown();
}
@Override public void failed(State from, Throwable failure) {
executor.shutdown();
}
}, directExecutor());
return executor;
}
/**
* Returns the name of this service. {@link AbstractScheduledService} may include the name in
* debugging output.
*
* @since 14.0
*/
protected String serviceName() {
return getClass().getSimpleName();
}
@Override public String toString() {
return serviceName() + " [" + state() + "]";
}
@Override public final boolean isRunning() {
return delegate.isRunning();
}
@Override public final State state() {
return delegate.state();
}
/**
* @since 13.0
*/
@Override public final void addListener(Listener listener, Executor executor) {
delegate.addListener(listener, executor);
}
/**
* @since 14.0
*/
@Override public final Throwable failureCause() {
return delegate.failureCause();
}
/**
* @since 15.0
*/
@Override public final Service startAsync() {
delegate.startAsync();
return this;
}
/**
* @since 15.0
*/
@Override public final Service stopAsync() {
delegate.stopAsync();
return this;
}
/**
* @since 15.0
*/
@Override public final void awaitRunning() {
delegate.awaitRunning();
}
/**
* @since 15.0
*/
@Override public final void awaitRunning(long timeout, TimeUnit unit) throws TimeoutException {
delegate.awaitRunning(timeout, unit);
}
/**
* @since 15.0
*/
@Override public final void awaitTerminated() {
delegate.awaitTerminated();
}
/**
* @since 15.0
*/
@Override public final void awaitTerminated(long timeout, TimeUnit unit) throws TimeoutException {
delegate.awaitTerminated(timeout, unit);
}
/**
* A {@link Scheduler} that provides a convenient way for the {@link AbstractScheduledService} to
* use a dynamically changing schedule. After every execution of the task, assuming it hasn't
* been cancelled, the {@link #getNextSchedule} method will be called.
*
* @author Luke Sandberg
* @since 11.0
*/
@Beta
public abstract static class CustomScheduler extends Scheduler {
/**
* A callable class that can reschedule itself using a {@link CustomScheduler}.
*/
private class ReschedulableCallable extends ForwardingFuture 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;
}
}