Класс инструмента Scheduled Task Timer в Java предоставляет функцию таймера или запланированного задания для выполнения задач в указанное время или интервал времени, но поскольку класс инструмента Timer реализован не в виде пула пула, а в виде очереди Он управляет потоками, поэтому эффективность работы низкая в случае высокого параллелизма.После версии JDK 1.5 объект ScheduledExecutorService предоставляется для решения проблем производительности, связанных с эффективностью и синхронизацией задач.
В этой статье мы в основном обсуждаем навыки использования ScheduledExecutorService и некоторые распространенные методы работы пула потоков.В последующих статьях будет продолжено подробное обсуждение исполнителей.
Класс инструментов Executors предоставляет два часто используемых ScheduledThreadPoolExecutors.
Эти два общих SCHEDULEDTHREADPOLEXECUTOR: SINGLETHREADSCHEDULEXECUTOR (пул с одним потоком), ScheduledthreadPool (пул с фиксированным номером потока), ниже приведен исходный код, соответствующий исполнителям.
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));
}
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory arg) {
return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1, arg));
}
public static ScheduledExecutorService newScheduledThreadPool(int arg) {
return new ScheduledThreadPoolExecutor(arg);
}
public static ScheduledExecutorService newScheduledThreadPool(int arg, ThreadFactory arg0) {
return new ScheduledThreadPoolExecutor(arg, arg0);
}
ScheduledExecutorService — это интерфейс, который наследуется от ExecutorService, поддерживает все функции пула потоков, а также предоставляет четыре основных метода для планирования запланированных задач.
Ниже я расскажу об использовании этих четырех методов и некоторых часто используемых методов пула потоков:
1. график работы
Планирование с задержкой выполняется только один раз, а возвращаемое значение представляет собой объект, который реализует интерфейс Future.Вы можете вызвать метод Future.get() для блокировки до тех пор, пока задача не будет выполнена.
/**
* 创建并执行在给定延迟后启用的一次性操作
*
* @param command 要执行的任务
* @param delay 从现在开始延迟执行的时间
* @param unit 延时参数的时间单位
* @return 表示任务等待完成，并且其的ScheduledFuture get()方法将返回 null
* @throws RejectedExecutionException 如果任务无法安排执行
* @throws NullPointerException 如果命令为空
*/
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
Пример запускаемого расписания
public static void main(String[] args) throws ExecutionException, InterruptedException {
ScheduledExecutorService scheduled = Executors.newSingleThreadScheduledExecutor();
ScheduledFuture<?> future = scheduled.schedule(() -> {
try {
System.out.println("开始执行任务");
TimeUnit.SECONDS.sleep(3);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("执行完毕");
}, 1000, TimeUnit.MILLISECONDS);
System.out.println("阻塞开始");
System.out.println(future.get() + "");
System.out.println("阻塞结束");
}
Результат выполнения следующий:
Блокировка начинается запустить задачу Законченный нулевой конец блокировки
расписание может выполняться, этот метод не предоставляет возвращаемого значения, поэтому вызов метода future.get() возвращает null
2. график вызова
Планирование с задержкой выполняется только один раз, а возвращаемое значение представляет собой объект, который реализует интерфейс Future.Вызовите метод Future.get() для блокировки до тех пор, пока задача не будет завершена, и можно будет получить возвращаемый результат.
/**
* 创建并执行在给定延迟后启用的ScheduledFuture
*
* @param callable 执行的功能
* @param delay 从现在开始延迟执行的时间
* @param unit 延迟参数的时间单位
* @param <V> the 可调用结果的类型
* @return一个可用于提取结果或取消的ScheduledFuture
* @throws RejectedExecutionException 如果该任务无法安排执行
* @throws NullPointerException 如果callable为空
*/
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
Пример использования Schedule Callable
public static void main(String[] args) throws ExecutionException, InterruptedException {
ScheduledExecutorService scheduled = Executors.newSingleThreadScheduledExecutor();
ScheduledFuture<String> future = scheduled.schedule(() -> {
try {
System.out.println("开始执行任务");
TimeUnit.SECONDS.sleep(3);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("执行完毕");
return "success";
}, 1000, TimeUnit.MILLISECONDS);
System.out.println("阻塞开始");
System.out.println(future.get() + "");
System.out.println("阻塞结束");
}
Результаты:
Блокировка начинается запустить задачу Законченный успех конец блокировки
Вызываемое расписание имеет возвращаемое значение, полученное через future.get()
3. расписание по фиксированной скорости
Создать и выполнить периодическое действие через заданную начальную задержку, то есть оно начнет выполняться после initialDelay, затем следующая задача выполняется через initialDelay+period, затем выполняется через initialDelay + 2 * period и так далее, т.е. есть задержка только при выполнении первой задачи.
/**
* @param command 要执行的任务
* @param initialDelay 首次执行的延迟时间
* @param period 连续执行之间的周期
* @param unit initialDelay和period参数的时间单位
* @return 一个ScheduledFuture代表待完成的任务，其 get()方法将在取消时抛出异常
* @throws RejectedExecutionException 如果任务无法安排执行
* @throws NullPointerException 如果命令为空
* @throws IllegalArgumentException 如果period小于或等于零
*/
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
Пример использования ScheduleAtFixedRate
public static void main(String[] args) throws ExecutionException, InterruptedException {
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(5);
ScheduledFuture<?> future = scheduled.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
System.out.println("开始执行任务");
TimeUnit.SECONDS.sleep(3);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("执行完毕");
}
}, 1000L, 1000L, TimeUnit.MILLISECONDS);
System.out.println("阻塞开始");
System.out.println(future.get() + "");
System.out.println("阻塞结束");
}
Результат печати следующий:
Блокировка начинается запустить задачу Законченный запустить задачу Законченный запустить задачу Законченный ....
4. расписание с фиксированной задержкой
Создает и выполняет периодическое действие, которое сначала включается после заданной начальной задержки, впоследствии существует заданная задержка между завершением каждого выполнения и началом следующего выполнения, т.е. общее время равно (initialDelay + период) * n
/**
* @param command 要执行的任务
* @param initialDelay 首次执行的延迟时间
* @param delay 一次执行终止和下一次执行开始之间的延迟
* @param unit initialDelay和delay参数的时间单位
* @return 表示挂起任务完成的ScheduledFuture，并且其get()方法在取消后将抛出异常
* @throws RejectedExecutionException 如果任务不能安排执行
* @throws NullPointerException 如果command为null
* @throws IllegalArgumentException 如果delay小于等于0
*/
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
Пример использования ScheduledWithFixedDelay
public static void main(String[] args) throws ExecutionException, InterruptedException {
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(5);
ScheduledFuture<?> future = scheduled.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
System.out.println("开始执行任务");
TimeUnit.SECONDS.sleep(3);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("执行完毕");
}
}, 1000L, 1000L, TimeUnit.MILLISECONDS);
System.out.println("阻塞开始");
System.out.println(future.get() + "");
System.out.println("阻塞结束");
}
Результат печати следующий:
Блокировка начинается запустить задачу Законченный запустить задачу Законченный запустить задачу Законченный ....
Разница между scheduleAtFixedRate и scheduleWithFixedDelay заключается в том, что scheduleAtFixedRate() — это фиксированная частота, а scheduleWithFixedDelay() — это фиксированная задержка. Фиксированная частота относительно времени начала выполнения задачи, а фиксированная задержка относительно времени окончания выполнения задачи, в этом их самое принципиальное отличие!
5. Отключение пула потоков, использование shutdown() и shutdownNow()
Два метода закрытия пула потоков: после закрытия пула все задачи, отправленные позже, будут отклонены.
Используйте shutdown(), чтобы использовать awaitTermination для ожидания завершения текущей задачи всеми потоками. Инициирует упорядоченное завершение выполнения ранее отправленных задач, но не принимает новые задачи.
Используйте shutdownNow(), чтобы попытаться остановить все выполняемые задачи, приостановить обработку ожидающих задач и вернуться к списку ожидающих задач. Для запуска попробуйте завершить поток, прервав его. Для задач, которые еще не запущены, они не будут выполнены.
class PrintThreadFactory implements ThreadFactory {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "PrintThreadFactory");
}
}
public static void main(String[] args) {
final AtomicInteger count = new AtomicInteger(0);
final CountDownLatch countDownLatch = new CountDownLatch(1);
ScheduledExecutorService schedule = Executors.newScheduledThreadPool(1, new PrintThreadFactory());
Runnable runnable = () -> {
System.out.println("print " + count.getAndIncrement());
if (count.get() == 3) {
countDownLatch.countDown();
System.out.println("任务继续...");
try {
Thread.sleep(3000L);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("任务结束");
}
};
schedule.scheduleAtFixedRate(runnable, 0L, 2L, TimeUnit.SECONDS);
try {
countDownLatch.await();
schedule.shutdown(); //平滑停止线程，不处理新任务，完成正在执行的任务
// schedule.shutdownNow(); // 尝试强制停止线程，让终止的线程去设置休眠会抛出异常
if (schedule.isShutdown()) {
System.out.println("Scheduled is shutdown");
}
if (schedule.awaitTermination(10L, TimeUnit.SECONDS)) {
System.out.println("termination");
}
} catch (Exception e) {
e.printStackTrace();
}
}
Отменить запущенные задачи через Future.cancel()
Параметр, полученный методом cancel(), является логическим значением.Передача значения true приведет к прерыванию потока для остановки задачи, а передача значения false позволит потоку выполняться в обычном режиме до завершения. Поскольку передача здесь false не прерывает поток, разве этот метод отмены не имеет смысла?
class PrintThreadFactory implements ThreadFactory {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "PrintThreadFactory");
}
}
public static void main(String[] args) {
final AtomicInteger count = new AtomicInteger(0);
final CountDownLatch countDownLatch = new CountDownLatch(1);
ScheduledExecutorService schedule = Executors.newScheduledThreadPool(1, new PrintThreadFactory());
Runnable runnable = () -> {
System.out.println("print " + count.getAndIncrement());
if (count.get() == 3) {
countDownLatch.countDown();
}
};
Future future = schedule.scheduleAtFixedRate(runnable, 0L, 2L, TimeUnit.SECONDS);
try {
countDownLatch.await();
future.cancel(true);
if (future.isCancelled()) {
System.out.println("is Cancelled");
}
} catch (Exception e) {
e.printStackTrace();
}
}
Проще говоря, передача параметра false может отменить только еще не запущенную задачу, если задача уже запущена, она будет продолжать выполняться. Итак, для уже запущенной задачи, если вы хотите ее остановить, вам нужно установить параметр метода отмены в значение true.
6. Использование параметра ScheduledThreadPoolExecutor
continueExistingPeriodicTasksAfterShutdown, допустимо для периодических задач, отправленных через scheduleAtFixedRate, scheduleWithFixedDelay. Значение по умолчанию — false, а значение true означает, что исполнитель продолжит выполнение отложенной задачи после вызова завершения работы;
Соответствующие методы get и set
void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value)
boolean getContinueExistingPeriodicTasksAfterShutdownPolicy()
executeExistingDelayedTasksAfterShutdown, действителен для отложенных задач, отправленных через метод schedule(), значение по умолчанию равно true, а установка значения false означает, что когда исполнитель вызывает завершение работы, он не будет продолжать выполнение существующих отложенных задач;
Соответствующие методы get и set
void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value)
boolean getExecuteExistingDelayedTasksAfterShutdownPolicy()
removeOnCancel, по умолчанию — false, установите значение true, чтобы удалить задачу выполнения из очереди;
Соответствующие методы get и set
void setRemoveOnCancelPolicy(boolean value)
boolean getRemoveOnCancelPolicy()
Пример использования
public static void main(String[] args) {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
Runnable runnable = () -> System.out.println(Thread.currentThread().getName() + "_1");
// ScheduledFuture<?> future = executor.schedule(runnable, 3, TimeUnit.SECONDS);
// 对于通过schedule()方法提交的延时任务有效，默认为true，设置为false表示当执行器调用shutdown后，不再继续执行现有延迟任务
// executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(true);
// System.out.println(executor.getQueue().size());
// // 默认为false，设置为true则从队列中删除执行任务
// executor.setRemoveOnCancelPolicy(false);
// future.cancel(true);
// System.out.println(executor.getQueue().size());
executor.scheduleAtFixedRate(runnable, 1L, 1L, TimeUnit.SECONDS);
// //对于通过scheduleAtFixedRate、scheduleWithFixedDelay 提交的周期任务有效 默认值为false，设置为true表示当执行器调用shutdown后，继续执行延时任务
executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(true);
executor.shutdown();
System.out.println("线程池停止");
}
7. Другое
Методы execute(), submit() с нулевой задержкой
Методы execute() и submit() были переписаны, а по сути вызывается метод schedule(); как видно из исходного кода ниже, задачи, отправляемые этими двумя методами, являются "задачами реального времени" с задержкой 0 ;
public void execute(Runnable arg0) {
this.schedule(arg0, 0L, TimeUnit.NANOSECONDS);
}
public Future<?> submit(Runnable arg0) {
return this.schedule(arg0, 0L, TimeUnit.NANOSECONDS);
}
Инкапсулирует класс инструментов пула запланированных задач.
Ниже приведен класс инструмента, инкапсулированный с использованием шаблона singleton.
public final class MyScheduledExecutor {
// 全局用于处理接收Future对象的集合
private ConcurrentHashMap<String, Future> futureMap = new ConcurrentHashMap<>();
// 计划执行任务
private ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(5);
private MyScheduledExecutor() {
}
// 设计为单例模式
private static final class InnerExecutorService {
private static final MyScheduledExecutor INSTANCE = new MyScheduledExecutor();
}
public static MyScheduledExecutor getInstance() {
return InnerExecutorService.INSTANCE;
}
public ConcurrentHashMap<String, Future> getFutureMap() {
return futureMap;
}
public void shutdown() {
executorService.shutdown();
}
/**
* 执行任务
* @param runnable {@code Runnable}
*/
public void execute(Runnable runnable) {
executorService.execute(runnable);
}
/**
* 执行延时任务
*
* @param runnable {@code Runnable}
* @param delay 延迟时间
* @param timeUnit 时间单位
*/
public void scheduler(Runnable runnable, long delay, TimeUnit timeUnit) {
executorService.schedule(runnable, delay, timeUnit);
}
/**
* 执行延时周期性任务scheduleAtFixedRate
*
* @param runnable {@code ScheduledExecutorService.JobRunnable}
* @param initialDelay 延迟时间
* @param period 周期时间
* @param timeUnit 时间单位
* @param <T> {@code ScheduledExecutorService.JobRunnable}
*/
public <T extends JobRunnable> void scheduleAtFixedRate(T runnable, long initialDelay, long period, TimeUnit timeUnit) {
Future future = executorService.scheduleAtFixedRate(runnable, initialDelay, period, timeUnit);
futureMap.put(runnable.getJobId(), future);
}
/**
* 执行延时周期性任务scheduleWithFixedDelay
*
* @param runnable {@code ScheduledExecutorService.JobRunnable}
* @param initialDelay 延迟时间
* @param period 周期时间
* @param timeUnit 时间单位
* @param <T> {@code ScheduledExecutorService.JobRunnable}
*/
public <T extends JobRunnable> void scheduleWithFixedDelay(T runnable, long initialDelay, long period, TimeUnit timeUnit) {
Future future = executorService.scheduleWithFixedDelay(runnable, initialDelay, period, timeUnit);
futureMap.put(runnable.getJobId(), future);
}
public static abstract class JobRunnable implements Runnable {
private String jobId;
public JobRunnable(String jobId) {
this.jobId = jobId;
}
public void terminal() {
try {
Future future = MyScheduledExecutor.getInstance().getFutureMap().remove(jobId);
future.cancel(true);
} finally {
System.out.println("jobId " + jobId + " had cancel");
}
}
public String getJobId() {
return jobId;
}
}
}
пример вызова
public static void main(String[] args) throws Exception {
MyScheduledExecutor service = MyScheduledExecutor.getInstance();
service.execute(() -> System.out.println("execute"));
// service.scheduler(new Runnable() {
// @Override
// public void run() {
// for (Map.Entry<String, Future> next : service.getFutureMap().entrySet()) {
// String key = next.getKey();
// int i = Integer.parseInt(key.substring(3));
// // 停止部分线程
// if (i % 2 == 0) {
// next.getValue().cancel(true);
// }
// }
// }
// }, 20, TimeUnit.SECONDS);
for (int i = 0; i < 5; i++) {
int num = new Random().nextInt(500);
service.scheduleAtFixedRate(new MyScheduledExecutor.JobRunnable("scheduleAtFixedRate" + num) {
@Override
public void run() {
System.out.println(num);
}
}, 10, 2, TimeUnit.SECONDS);
}
Thread.sleep(15000);
for (Map.Entry<String, Future> next : service.getFutureMap().entrySet()) {
String key = next.getKey();
int i = Integer.parseInt(key.substring(3));
// 停止部分线程
if (i % 2 == 0) {
next.getValue().cancel(true);
}
}
Thread.sleep(20000);
service.shutdown();
}
Суммировать
Следует отметить, что для периодических задач, выполняемых ScheduledExecutorService, если во время выполнения задачи возникнет исключение, ScheduledExecutorService прекратит выполнение задачи и не будет выполнять ее периодически. Поэтому, если вы хотите, чтобы задачи выполнялись периодически, перехватывайте все возможные исключения.
Если содержание статьи для вас полезно, обратите внимание на публичный номер优享JAVA（ID:YouXiangJAVA）, вас ждут другие технические галантереи, а список книг для программистов был тщательно подготовлен. с нетерпением ждем вашего визита!