Java +线程：并行处理线

Question

我想并行处理大量独立的线。在下面的代码中，我创建了一个包含 POOL_SIZE 行的 NUM_THREAD 个主题池。 每个线程都启动，然后我使用“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 util.concurrent API

Answer 3

实际上不鼓励直接使用线程 - 查看 java.util.concurrent 包，您会发现应该使用 ThreadPools 和 Futures 来代替。

Thread.join 并不意味着该线程等待其他线程，它意味着您的主线程等待列表中的线程之一死亡。在这种情况下，您的主线程等待最慢的工作线程完成。我认为这种方法没有问题。

Answer 4

是的，从某种意义上说，一个已完成的线程必须等待池中的同级线程：当一个线程完成时，它会停止，并且不会帮助其他线程更快地完成。更好的说法是，整个工作等待工作时间最长的线程。

这是因为每个线程只有一个任务。您最好创建比线程数量多得多的任务，并将它们全部放在一个队列中。让所有工作线程循环地从该队列中获取任务。那么所有线程的时间差将大致是执行一项任务的时间，由于任务很小，所以这个时间很小。

您可以自己启动工作线程池，也可以将每个任务包装在 Runnable 中并将它们提交到标准线程池 - 这没有什么区别。