Java: Как получить готовые потоки для задач загрузки из запущенных потоков

Question

Я работаю над многопоточным приложением с задачами, которые меняют время выполнения. Когда один поток заканчивается, есть ли способ, чтобы он мог взять на себя некоторые задачи из неподвижного потока?

Вот пример. Я запускаю свою программу с 5 потоками, и у каждого из них 50 задач. Когда заканчивается самый быстрый поток, другой поток по-прежнему выполняет 40 задач. Как я могу получить готовый поток для выполнения 20 задач из другого потока, поэтому каждый из них продолжает работать на 20 штук, вместо того, чтобы ждать, пока работающий поток завершит оставшиеся 40?

Answer 1

Использование ForkJoinPool

ForkJoinPool отличается от других видов ExecutorService главным образом в силу использования работы крадущую: все темы, в попытке бассейна, чтобы найти и выполнить подзадачи, созданные другими активными задачами (в конечном счете, блокирование в ожидании работа если нет есть). Это позволяет эффективно обрабатывать, когда большинство задач порождают другие подзадачи (как и большинство ForkJoinTasks). При установке asyncMode в true в конструкторах, ForkJoinPools также может быть подходящим для использования с задачами в стиле событий, которые никогда не соединяются.

Java 8 предоставляет еще один API в исполнителях

static ExecutorService newWorkStealingPool() 

Создает рабочую крадущую пул потоков, используя все доступные процессоры, как его целевой уровень параллелизма.

Посмотрите на эту igvtia article по Ilya Grigorik для получения более подробной информации.

Посмотрите на другие связанные с Java API параллельного @tutorials как ThreadPoolExecutor, ExecutorService и т.д.

Answer 2

Лучше использовать ThreadPoolExecutor. Он автоматически назначает задачи для бесплатных потоков.

Answer 3

Использование пулов потоков, которые создаются благодаря Executors класса:

ExecutorService es = Executors.newFixedThreadPool(5); 
List<Runnable> tasks = // create your 50 runnable 
List<Future<?>> futures = new ArrayList<>(tasks.size()); 
for(Runnable r : tasks) { 
    Future<?> f = es.submit(t); 
    futures.add(f); 
} 

документация объясняет довольно хорошо, как это работает, так что я рекомендую вам придать ему вид.

Answer 4

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

Рассмотрите - 2 потока имеют по 20 задач каждый, и вы хотите, чтобы второй поток выполнял задачи с первого, если он еще не закончен. Сравните это с тем, что 40 задач в очереди обслуживаются 2 потоками, что означает, что задачи всегда будут выполняться как можно быстрее без сложностей при попытке их перемещения между потоками.

Я не вижу логики в вопросе - если у вас есть группы задач, которые не могут быть многопоточными из-за проблем с секвенированием (это единственная причина, по которой я могу представить задачу, которая сама по себе является группой задачи в очередь), то у вас не может быть другого потока, обрабатывающего незавершенную обработку (потому что тогда последовательность всей группы будет нарушена).Если вам не требуется последовательная обработка, то бросьте все задачи в очередь и пусть они будут выполнены как можно скорее.

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