Политика ThreadPoolExecutor

Question

Я пытаюсь использовать ThreadPoolexecuteCutor, чтобы запланировать задачи, но в течение некоторых проблем с его политикой. Вот его заявленное поведение:

1. Если запущены меньше потоков COREPOOLSIZE, исполнитель всегда предпочитает добавлять новый поток, а не в очереди.
2. Если работает COREPOOLSIZE или более потоков, исполнитель всегда предпочитает вспомнить запрос, а не добавлять новый поток.
3. Если запрос не может быть введен в очередь, создается новый поток, если это не превысит максимальный пункт, в этом случае задача будет отклонена.

Поведение, которое я хочу, это:

1. так же, как указано выше
2. Если работает больше, чем COREPOOLSIZE, но меньше, чем максимампульсизация потоков, предпочитает добавлять новую тему через очередь, а также с помощью простоя едок по добавлению нового потока.
3. так же, как указано выше

В основном я не хочу, чтобы какие-либо задачи были отклонены; Я хочу, чтобы они были в очереди в неограниченной очереди. Но я хочу иметь максимума импульсизма. Если я использую неограниченную очередь, он никогда не генерирует потоки после того, как он попадает в CoREsize. Если я использую ограниченную очередь, она отклоняет задачи. Есть ли вокруг этого?

Я думаю, что сейчас я думаю, управляет ThreadPoolexecuteCutor на синхронном море, но не кормит задачи непосредственно к нему - вместо этого кормить их до отдельной неограниченной LinkedBlockingQueue. Затем еще одна нить питается от ссылкиBlockingceueue в исполнителя, и если кто-то отклоняется, он просто снова пытается, пока он не отклонен. Это похоже на боль и немного взлома, хотя - есть ли более чистый способ сделать это?

Answer 1

Ваш корпус использования является обычным, полностью законным и, к сожалению, сложнее, чем можно ожидать. Для фоновой информации вы можете прочитать Это обсуждение и найти указатель на решение (также упоминается в потоке) здесь. Отказ Решение Shay работает нормально.

Как правило, я был бы немного настороженным из неограниченных очередей; Обычно лучше иметь явный контроль входящих потоков, который изящно ухудшается и регулирует соотношение тока / оставшейся работы, чтобы не сокрушить либо производителя, либо потребителя.

Answer 2

Вероятно, это не нужно для микро-управления пулом нитей как запрашиваемого.

Кэшированный пул резьбы повторно использует холостые нити, а также позволяя потенциально неограниченным параллельным потокам. Это, конечно, может привести к запуску производительности ухудшения от переключения контекста в течение разрывных периодов.

Executors.newCachedThreadPool();

Лучший вариант - разместить предел на общее количество потоков, сбрасывая понятие для обеспечения использования холостых резьбовых потоков. Изменения конфигурации были бы:

corePoolSize = maximumPoolSize = N;
allowCoreThreadTimeOut(true);
setKeepAliveTime(aReasonableTimeDuration, TimeUnit.SECONDS);

Рассуждения над этим сценарием, если исполнитель меньше, чем corePoolSize Темы, чем не должны быть очень заняты. Если система не очень занята, то в спиннении мало вреда от новой нити. Это приведет к этому ThreadPoolExecutor Чтобы всегда создать нового работника, если он находится под максимальным количеством работников. Только когда максимальное количество рабочих «работает», будут лидеры рабочие, ожидающие задачи. Если работник ждет aReasonableTimeDuration Без задачи, то это разрешено прекращать. Используя разумные пределы для размера бассейна (ведь в конце концов, есть только многие процессоры), а достаточно большой тайм-аут (чтобы сохранить нити без необходимости прекращения), желаемые преимущества, вероятно, будут видны.

Окончательный вариант - это взлома. В основном ThreadPoolExecutor внутренне использует BlockingQueue.offer определить, имеет ли очередь емкости. Пользовательская реализация BlockingQueue всегда мог отклонить offer пытаться. Когда ThreadPoolExecutor не удается offer Задача в очередь, она попытается сделать новый работник. Если новый работник не может быть создан, RejectedExecutionHandler будет называться. В этот момент обычай RejectedExecutionHandler может заставить А. put в таможне BlockingQueue.

/** Hackish BlockingQueue Implementation tightly coupled to ThreadPoolexecutor implementation details. */
class ThreadPoolHackyBlockingQueue<T> implements BlockingQueue<T>, RejectedExecutionHandler {
    BlockingQueue<T> delegate;

    public boolean offer(T item) {
        return false;
    }

    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        delegate.put(r);
    }

    //.... delegate methods
}

Answer 3

Просто установить corePoolsize = maximumPoolSize и используйте неограниченную очередь?

В вашем списке очков 1 исключает 2, поскольку corePoolSize всегда будет меньше или равна, чем maximumPoolSize.

Редактировать

Есть еще что-то несовместимое между тем, что вы хотите, и какой TPE вам предложит вам.

Если у вас неограниченная очередь, maximumPoolSize игнорируется так, как вы наблюдали, не более corePoolSize Темы когда-либо будут созданы и использованы.

Итак, опять же, если вы берете corePoolsize = maximumPoolSize С неограниченной очередью у вас есть то, что вы хотите, нет?

Answer 4

Вы бы искали что-то более похоже на кэшированную резьбу пула?

http://download.orcle.com/javase/1.5.0/docs/api/java/util/concurrent/executors.html#newcachedThreadpool ()