Как остановить Runnable, запланированный для повторного выполнения после определенного количества выполнений
Ситуация
У меня есть Runnable. У меня есть класс, который планирует этот Runnable для выполнения с использованием ScheduledExecutorService с scheduleWithFixedDelay .
Цель
Я хочу изменить этот класс, чтобы запланировать Runnable для выполнения с фиксированной задержкой или на неопределенный срок, или , пока он не будет выполнен определенное количество раз. в зависимости от некоторого параметра, который передается конструктору.
Если возможно, я бы хотел использовать тот же Runnable, поскольку это концептуально то же самое, что должно быть «запущено».
Возможные подходы
Подход № 1
Имеют два Runnables, один, который отменяет расписание после ряда выполнений (которое ведет счет), и другой, который не делает:
public class MyClass{
private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
public enum Mode{
INDEFINITE, FIXED_NO_OF_TIMES
}
public MyClass(Mode mode){
if(mode == Mode.INDEFINITE){
scheduler.scheduleWithFixedDelay(new DoSomethingTask(), 0, 100, TimeUnit.MILLISECONDS);
}else if(mode == Mode.FIXED_NO_OF_TIMES){
scheduler.scheduleWithFixedDelay(new DoSomethingNTimesTask(), 0, 100, TimeUnit.MILLISECONDS);
}
}
private class DoSomethingTask implements Runnable{
@Override
public void run(){
doSomething();
}
}
private class DoSomethingNTimesTask implements Runnable{
private int count = 0;
@Override
public void run(){
doSomething();
count++;
if(count > 42){
// Cancel the scheduling.
// Can you do this inside the run method, presumably using
// the Future returned by the schedule method? Is it a good idea?
}
}
}
private void doSomething(){
// do something
}
}
Я бы предпочел иметь только один Runnable для выполнения метода doSomething. Привязка планирования к Runnable кажется неправильной. Что вы думаете об этом?
Подход № 2
Есть один Runnable для выполнения кода, который мы хотим периодически запускать. Иметь отдельный запланированный runnable, который проверяет, сколько раз был запущен первый Runnable, и отменяет его при достижении определенной суммы. Это может быть неточным, так как это будет асинхронным. Это кажется немного громоздким. Что вы думаете об этом?
Подход № 3
Расширьте ScheduledExecutorService и добавьте метод scheduleWithFixedDelayNTimes. Возможно, такой класс уже существует? В настоящее время я использую Executors.newSingleThreadScheduledExecutor(); для получения своего экземпляра ScheduledExecutorService. Я бы, вероятно, должен был реализовать аналогичную функциональность для создания экземпляра расширенного ScheduledExecutorService. Это может быть сложно. Что вы думаете по этому поводу?
Нет подхода к планировщику [Edit]
Я не смог использовать планировщик. Вместо этого я мог бы получить что-то вроде:
for(int i = 0; i < numTimesToRun; i++){
doSomething();
Thread.sleep(delay);
}
И запустить это в каком-то потоке. Что вы думаете об этом? Вы потенциально можете по-прежнему использовать runnable и напрямую вызывать метод run.
Любые предложения приветствуются. Я ищу дебаты, чтобы найти «лучший практический» способ достижения моей цели.
2 ответа
Для вариантов использования как опрос до определенного тайм-аута мы можем приблизиться с простым решением с помощью Future.get().
/* Define task */
public class Poll implements Runnable {
@Override
public void run() {
// Polling logic
}
}
/* Create executor service */
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
/* Schedule task - poll every 500ms */
ScheduledFuture<?> future = executorService.scheduleAtFixedRate(new Poll(), 0, 500, TimeUnit.MILLISECONDS);
/* Wait till 60 sec timeout */
try {
future.get(60, TimeUnit.SECONDS);
} catch (TimeoutException e) {
scheduledFuture.cancel(false);
// Take action on timeout
}
Я искал точно ту же самую функциональность и выбрал org.springframework.scheduling.Trigger.
Ниже полный тест рабочий пример (извините если слишком много лавинной рассылки в коде) applicationContext.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context/ http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/util/ http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd">
<bean id="blockingTasksScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
<property name="poolSize" value="10" />
</bean>
<task:scheduler id="deftaskScheduler" pool-size="10" />
</beans>
JAVA
package com.alz.springTests.schedulerTest;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
import java.util.Date;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
public class ScheduledTest {
private static ApplicationContext applicationContext;
private static TaskScheduler taskScheduler;
private static final class SelfCancelableTask implements Runnable, Trigger {
Date creationTime = new Date();
AtomicInteger counter = new AtomicInteger(0);
private volatile boolean shouldStop = false;
private int repeatInterval = 3; //seconds
@Override
public void run() {
log("task: run started");
// simulate "doing job" started
int sleepTimeMs = ThreadLocalRandom.current().nextInt(500, 2000+1);
log("will sleep " + sleepTimeMs + " ms");
try {
Thread.sleep(sleepTimeMs);
} catch (InterruptedException e) {
e.printStackTrace();
}
// "doing job" finished
int i = counter.incrementAndGet();
if (i > 5) { //cancel myself
logErr("Attempts exceeded, will mark as shouldStop");
shouldStop = true;
} else {
log("task: executing cycle #"+i);
}
}
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
log("nextExecutionTime: triggerContext.lastActualExecutionTime() " + triggerContext.lastActualExecutionTime());
log("nextExecutionTime: triggerContext.lastCompletionTime() " + triggerContext.lastCompletionTime());
log("nextExecutionTime: triggerContext.lastScheduledExecutionTime() " + triggerContext.lastScheduledExecutionTime());
if (shouldStop)
return null;
if (triggerContext.lastCompletionTime() == null) {
LocalDateTime ldt = creationTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime().plus(repeatInterval, ChronoUnit.SECONDS);
return Date.from(ldt.atZone(ZoneId.systemDefault()).toInstant());
} else {
LocalDateTime ldt = triggerContext.lastCompletionTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime().plus(repeatInterval, ChronoUnit.SECONDS);
return Date.from(ldt.atZone(ZoneId.systemDefault()).toInstant());
}
}
}
private static void log(String log) {
System.out.printf("%s [%s] %s\r\n", LocalDateTime.now(), Thread.currentThread(), log);
}
private static void logErr(String log) {
System.err.printf("%s [%s] %s\r\n", LocalDateTime.now(), Thread.currentThread(), log);
}
public static void main(String[] args) {
log("main: Stated...");
applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
taskScheduler = (TaskScheduler) applicationContext.getBean("blockingTasksScheduler");
ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = ((ThreadPoolTaskScheduler)taskScheduler).getScheduledThreadPoolExecutor();
SelfCancelableTask selfCancelableTask = new SelfCancelableTask();
taskScheduler.schedule(selfCancelableTask, selfCancelableTask);
int waitAttempts = 0;
while (waitAttempts < 30) {
log("scheduledPool pending tasks: " + scheduledThreadPoolExecutor.getQueue().size());
try {
Thread.sleep(1*1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
waitAttempts++;
}
log("main: Done!");
}
}