Java + Threads: параллельная обработка строк

Question

Я хочу обрабатывать большое количество независимых строк параллельно. В следующем коде я создаю пул NUM_THREAD Theads, содержащий POOL_SIZE строк. Каждый поток запускается, и я затем жду каждого потока, используя 'join'.

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

Как правильно реализовать этот код? Какие классы мне следует использовать?

Спасибо!

class FasterBin extends Thread
    {
    private List<String> dataRows=new ArrayList<String>();
    private Object result=null;
    @Override
    public void run()
        {
        for(String s:dataRows)
            {
            //Process item here (....)
            }
        }
    }


(...)

List<FasterBin> threads=new Vector<FasterBin>();
String line;
Iterator<String> iter=(...);
for(;;)
    {
    while(threads.size()< NUM_THREAD)
        {
        FasterBin bin=new FasterBin();
        while(
            bin.dataRows.size() < POOL_SIZE &&
            iter.hasNext()
            )
            {
            nRow++;
            bin.dataRows.add(iter.next());
            }
        if(bin.dataRows.isEmpty()) break;
        threads.add(bin);
        }
    if(threads.isEmpty()) break;


    for(FasterBin t:threads)
        {
        t.start();
        }
    for(FasterBin t:threads)
        {
        t.join();
        }
    for(FasterBin t:threads)
        {
        save(t.result);// ## do something with the result (save into a db etc...)
        }

    threads.clear();
    }

finally
    {
    while(!threads.isEmpty())
        {

        FasterBin b=threads.remove(threads.size()-1);
        try     {
            b.interrupt();
            }
        catch (Exception e)
            {
            }
        }
    }

Answer 1

Делай НЕ делай все это сам!Очень трудно получить 1) надежный и 2) правильный.

Вместо этого переписать свои вещи, чтобы создать множество Runnables или Callables и использовать подходящий ExecutorService, чтобы получить Executor для обработки их с желаемым поведением.

Обратите внимание, что это остается внутри текущей JVM.Если у вас имеется более одной JVM (на нескольких машинах), я бы порекомендовал открыть новый вопрос.

Answer 2

java.util.concurrent.ThreadPoolExecutor.

        ThreadPoolExecutor  x=new ScheduledThreadPoolExecutor(10);
        x.execute(runnable);

См. Это для обзора: Java API для util.concurrent

Answer 3

Прямое использование Threads на самом деле не рекомендуется - посмотрите на пакет java.util.concurrent, там вы найдете ThreadPools и Futures, которые следует использовать вместо этого.

Thread.join не означает, что Поток ждет других, это означает, что ваш основной Поток ожидает, чтобы один из Потоков в списке умер. В этом случае ваш основной поток ожидает окончания работы самого медленного рабочего потока. Я не вижу проблем с этим подходом.

Answer 4

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

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

Вы можете запустить пул рабочих потоков самостоятельно, или вы можете обернуть каждую задачу в Runnable и отправить их в стандартный пул потоков - это не имеет значения.