我通过创建一个可以“限制”任务的执行器,以通用的方式解决了这个问题。
任务(Runnables)被放置在
DelayQueue
中,从那里调度线程取出并执行它们。最新调度的任务也被记在一个变量中,因此如果调度程序从队列中检索到一个新任务,他会检查是否这是最新计划的任务。如果是,则执行它;否则跳过它。
我使用一个字符串标识符来检查哪些任务被认为属于一个“限制”。
以下是代码,其中还包括普通调度功能,但您可以在其中查看关键部分。
package org.uilib.util;
import com.google.common.collect.Maps;
import java.util.Map;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public final class SmartExecutor implements Throttle, Executor {
private static final Logger L = LoggerFactory.getLogger(SmartExecutor.class);
private final ExecutorService executor = Executors.newCachedThreadPool();
private final DelayQueue<DelayedRunnable> taskQueue = new DelayQueue<DelayedRunnable>();
private final Map<String, ThrottledRunnable> throttledTasks = Maps.newHashMap();
public SmartExecutor() {
this.executor.execute(new Scheduler());
}
@Override
public void execute(final Runnable runnable) {
this.executor.execute(runnable);
}
public void schedule(final long delay, final TimeUnit timeUnit, final Runnable runnable) {
this.taskQueue.put(new DelayedRunnable(runnable, delay, timeUnit));
}
public void scheduleAtFixedRate(final long period, final TimeUnit timeUnit, final Runnable runnable) {
this.taskQueue.put(new RepeatingRunnable(runnable, period, timeUnit));
}
public void shutdown() {
this.executor.shutdownNow();
}
@Override
public void throttle(final String throttleName, final long delay, final TimeUnit timeUnit, final Runnable runnable) {
final ThrottledRunnable thrRunnable = new ThrottledRunnable(runnable, throttleName, delay, timeUnit);
this.throttledTasks.put(throttleName, thrRunnable);
this.taskQueue.put(thrRunnable);
}
private static class DelayedRunnable implements Delayed, Runnable {
protected final Runnable runnable;
private final long endOfDelay;
public DelayedRunnable(final Runnable runnable, final long delay, final TimeUnit delayUnit) {
this.runnable = runnable;
this.endOfDelay = delayUnit.toMillis(delay) + System.currentTimeMillis();
}
@Override
public int compareTo(final Delayed other) {
final Long delay1 = this.getDelay(TimeUnit.MILLISECONDS);
final Long delay2 = other.getDelay(TimeUnit.MILLISECONDS);
return delay1.compareTo(delay2);
}
@Override
public long getDelay(final TimeUnit unit) {
return unit.convert(this.endOfDelay - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public void run() {
this.runnable.run();
}
}
private static final class RepeatingRunnable extends DelayedRunnable {
private final long periodInMillis;
public RepeatingRunnable(final Runnable runnable, final long period, final TimeUnit delayUnit) {
super(runnable, period, delayUnit);
this.periodInMillis = delayUnit.convert(period, TimeUnit.MILLISECONDS);
}
public RepeatingRunnable reschedule() {
return new RepeatingRunnable(this.runnable, this.periodInMillis, TimeUnit.MILLISECONDS);
}
}
private final class Scheduler implements Runnable {
@Override
public void run() {
while (true) {
try {
final DelayedRunnable task = SmartExecutor.this.taskQueue.take();
if (task instanceof RepeatingRunnable) {
SmartExecutor.this.executor.execute(
new Runnable() {
@Override
public void run() {
task.run();
SmartExecutor.this.taskQueue.put(((RepeatingRunnable) task).reschedule());
}
});
} else if (task instanceof ThrottledRunnable) {
final ThrottledRunnable thrTask = (ThrottledRunnable) task;
if (SmartExecutor.this.throttledTasks.get(thrTask.getThrottleName()) == thrTask) {
SmartExecutor.this.executor.execute(task);
}
} else {
SmartExecutor.this.executor.execute(task);
}
} catch (final InterruptedException e) {
SmartExecutor.L.debug("scheduler interrupted (shutting down)");
return;
}
}
}
}
private static final class ThrottledRunnable extends DelayedRunnable {
private final String throttleName;
public ThrottledRunnable(final Runnable runnable, final String throttleName, final long period,
final TimeUnit delayUnit) {
super(runnable, period, delayUnit);
this.throttleName = throttleName;
}
public String getThrottleName() {
return this.throttleName;
}
}
}
if ((lastEvent + 1000) < System.currentTimeMillis())
而不是>
。 - PeterVermont