我想编写一段可重用的代码,以便在将任务提交到执行程序服务时允许等待条件。如果队列中的任务过多,则有很多巧妙的阻塞方式的实现,例如这里
我需要一个执行器来评估所有等待的线程,每次任务完成。为了决定是否允许在 atm 提交任务,必须考虑所有活动任务的当前状态。我想出了以下解决方案,它不必针对多个提交者或同时执行的高级任务进行扩展。
问题:以下代码可以安全使用,还是我遗漏了一些缺陷?实现aquireAccess
方法的人ConditionEvaluator<T>
必须确保查询线程状态的方式是线程安全的,但实现者不需要保护对 activeTasks 集合的迭代。这是代码:
public class BlockingExecutor<T extends Runnable> {
private final Executor executor;
private final ConditionEvaluator<T> evaluator;
final ReentrantLock lock = new ReentrantLock();
final Condition condition = this.lock.newCondition();
final LinkedList<T> active = new LinkedList<T>();
private final long reevaluateTime;
private final TimeUnit reevaluateTimeUnit;
public BlockingExecutor(Executor executor, ConditionEvaluator<T> evaluator) {
this.evaluator = evaluator;
this.executor = executor;
this.reevaluateTimeUnit = null;
this.reevaluateTime = 0;
}
public BlockingExecutor(Executor executor, ConditionEvaluator<T> evaluator, long reevaluateTime, TimeUnit reevaluateTimeUnit) {
this.evaluator = evaluator;
this.executor = executor;
this.reevaluateTime = reevaluateTime;
this.reevaluateTimeUnit = reevaluateTimeUnit;
}
public void submitTask(final T task) throws InterruptedException {
this.lock.lock();
try {
do{
if (this.reevaluateTimeUnit == null) {
this.condition.await(this.reevaluateTime, this.reevaluateTimeUnit);
} else {
this.condition.await();
}
}while(!this.evaluator.aquireAccess(this.active, task));
this.active.add(task);
this.executor.execute(new Runnable() {
@Override
public void run() {
try {
task.run();
} finally {
BlockingExecutor.this.lock.lock();
try{
BlockingExecutor.this.active.remove(task);
BlockingExecutor.this.condition.signalAll();
}finally{
BlockingExecutor.this.lock.unlock();
}
}
}
});
} finally {
this.lock.unlock();
}
}
}
public interface ConditionEvaluator<T extends Runnable> {
public boolean aquireAccess(Collection<T> activeList,T task);
}
问题:代码可以改进吗?