0

我想编写一段可重用的代码,以便在将任务提交到执行程序服务时允许等待条件。如果队列中的任务过多,则有很多巧妙的阻塞方式的实现,例如这里

我需要一个执行器来评估所有等待的线程,每次任务完成。为了决定是否允许在 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);   
}

问题:代码可以改进吗?

4

0 回答 0