указание проблемы ThreadPoolExecutor - PullRequest
Новая
       7

указание проблемы ThreadPoolExecutor

5 голосов
/ 17 сентября 2009

Есть ли способ создать Executor, у которого всегда будет как минимум 5 потоков, максимум 20 потоков и неограниченная очередь для задач (то есть ни одна задача не отклоняется)

Я попробовал новый ThreadPoolExecutor(5, 20, 60L, TimeUnit.SECONDS, queue) со всеми возможными вариантами очереди: 

new LinkedBlockingQueue() // never runs more than 5 threads
new LinkedBlockingQueue(1000000) // runs more than 5 threads, only when there is more than 1000000 tasks waiting
new ArrayBlockingQueue(1000000) // runs more than 5 threads, only when there is more than 1000000 tasks waiting
new SynchronousQueue() // no tasks can wait, after 20, they are rejected

и никто не работал так, как хотел.

Ответы [ 3 ]

5 голосов
/ 17 сентября 2009

Может быть, что-то подобное подойдет вам? Я просто взбил его, так что, пожалуйста, тыкайте в него. По сути, он реализует пул потоков переполнения, который используется для подачи базового ThreadPoolExecutor

Я вижу два основных недостатка:

  • Отсутствие возвращенного объекта Future на submit(). Но, может быть, это не проблема для вас.
  • Вторичная очередь будет очищаться в ThreadPoolExecutor только при отправке заданий. Должно быть элегантное решение, но я пока не вижу его. Если вы знаете, что в StusMagicExecutor будет постоянный поток задач, то это может не быть проблемой. («Может» - ключевое слово.) Возможно, вы можете сделать так, чтобы отправленные вами задания ткнули в StusMagicExecutor после их завершения?

Магический Исполнитель Стю: 

public class StusMagicExecutor extends ThreadPoolExecutor {
    private BlockingQueue<Runnable> secondaryQueue = new LinkedBlockingQueue<Runnable>();  //capacity is Integer.MAX_VALUE.

    public StusMagicExecutor() {
        super(5, 20, 60L, SECONDS, new SynchronousQueue<Runnable>(true), new RejectionHandler());  
    }
    public void queueRejectedTask(Runnable task) {
        try {
            secondaryQueue.put(task);
        } catch (InterruptedException e) {
            // do something
        }
    }
    public Future submit(Runnable newTask) {
        //drain secondary queue as rejection handler populates it
        Collection<Runnable> tasks = new ArrayList<Runnable>();
        secondaryQueue.drainTo(tasks);

        tasks.add(newTask);

        for (Runnable task : tasks)
             super.submit(task);

        return null; //does not return a future!
    }
}

class RejectionHandler implements RejectedExecutionHandler {
    public void rejectedExecution(Runnable runnable, ThreadPoolExecutor executor) {
        ((StusMagicExecutor)executor).queueRejectedTask(runnable);
    }
}
1 голос
/ 31 марта 2010

Я думаю, что эта проблема является недостатком класса и вводит в заблуждение, учитывая комбинации параметров конструктора. Вот решение, взятое из внутреннего ThreadPoolExecutor SwingWorker, которое я превратил в класс верхнего уровня. У него нет минимума, но, по крайней мере, используется верхняя граница. Единственное, чего я не знаю, это то, какой удар по производительности вы получаете от выполнения блокировки. 

public class BoundedThreadPoolExecutor extends ThreadPoolExecutor {
    private final ReentrantLock pauseLock = new ReentrantLock();
    private final Condition unpaused = pauseLock.newCondition();
    private boolean isPaused = false;
    private final ReentrantLock executeLock = new ReentrantLock();

    public BoundedThreadPoolExecutor(int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(0, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public BoundedThreadPoolExecutor(int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
        BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(0, maximumPoolSize, keepAliveTime, unit, workQueue,
                threadFactory);
    }

    public BoundedThreadPoolExecutor(int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(0, maximumPoolSize, keepAliveTime, unit, workQueue,
                handler);
    }

    public BoundedThreadPoolExecutor(int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
            RejectedExecutionHandler handler) {
        super(0, maximumPoolSize, keepAliveTime, unit, workQueue,
                threadFactory, handler);
    }

    @Override
    public void execute(Runnable command) {
        executeLock.lock();
        try {
            pauseLock.lock();
            try {
                isPaused = true;
            } finally {
                pauseLock.unlock();
            }
            setCorePoolSize(getMaximumPoolSize());
            super.execute(command);
            setCorePoolSize(0);
            pauseLock.lock();
            try {
                isPaused = false;
                unpaused.signalAll();
            } finally {
                pauseLock.unlock();
            }
        } finally {
            executeLock.unlock();
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        pauseLock.lock();
        try {
            while (isPaused) {
                unpaused.await();
            }
        } catch (InterruptedException ignore) {

        } finally {
            pauseLock.unlock();
        }
    }
}
1 голос
/ 17 сентября 2009

В javadocs для ThreadPoolExecutor достаточно ясно, что после создания потоков corePoolSize новые потоки будут создаваться только после заполнения очереди. Поэтому, если вы установите core на 5 и max на 20, вы никогда не получите желаемого поведения.

Однако, если вы установите для core и max значение 20, задачи будут добавляться в очередь, только если все 20 потоков заняты. Конечно, это делает ваше требование «минимум 5 потоков» немного бессмысленным, так как все 20 будут поддерживаться (в любом случае, пока они не отключатся).

...