强制多个线程在可用时使用多个 CPU答案

【问题标题】：Forcing multiple threads to use multiple CPUs when they are available强制多个线程在可用时使用多个 CPU
【发布时间】：2010-11-16 10:07:55
【问题描述】：

我正在编写一个使用大量 CPU 的 Java 程序，因为它的工作性质。但是，其中很多可以并行运行，并且我已经使我的程序成为多线程的。当我运行它时，它似乎只使用一个 CPU，直到它需要更多然后它使用另一个 CPU - 我可以在 Java 中做些什么来强制不同的线程在不同的内核/CPU 上运行？

【问题讨论】：

我不确定你在问什么，现在我想了想。您是在问 (a) 如何让它在多个线程中运行 (b) 为什么多线程代码使用的内核不超过一个内核或 (c) 为什么 CPU 负载不均匀分布？
您的应用程序没有足够的可独立运行的任务来一次使用多个 CPU。这个问题不太可能出现在您的操作系统中，因为这些问题已经经过数百万多年来的测试。您应该再次查看您的程序，以查看您希望同时运行哪些任务，同时尝试确定是什么阻止了这种情况的发生。
Peter，如果一个应用程序运行两个线程，那么它就有足够的东西在多个内核上运行。即使所有其他线程所做的只是启动和终止，仍然有工作负载可在第二个核心上运行。 -- 仅仅因为每个线程目前似乎没有高工作负载，就将单个内核分配给多个线程，这是适得其反的。同步到一些非 CPU 工作负载的想法怎么样（例如一个通用的附加计算板，表明它已经完成了它的工作负载）。线程亲和性非常很重要！ Java 应该支持这一点。

标签： java concurrency multithreading multicore

【解决方案1】：

您可以使用来自Executors 的以下 API 和 Java 8 版本

public static ExecutorService newWorkStealingPool()

使用所有可用处理器作为目标并行级别创建一个工作窃取线程池。

由于工作窃取机制，空闲线程从繁忙线程的任务队列中窃取任务，整体吞吐量会增加。

从grepcode，newWorkStealingPool的实现如下

/**
     * Creates a work-stealing thread pool using all
     * {@link Runtime#availableProcessors available processors}
     * as its target parallelism level.
     * @return the newly created thread pool
     * @see #newWorkStealingPool(int)
     * @since 1.8
     */
    public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

【讨论】：

【解决方案2】：

之前在Why does this Java code not utilize all CPU cores? 中已经提到过JVM 性能调优。请注意，这仅适用于 JVM，因此您的应用程序必须已经在使用线程（并且或多或少“正确”）：

http://ch.sun.com/sunnews/events/2009/apr/adworkshop/pdf/5-1-Java-Performance.pdf

【讨论】：

+1 供参考。 PDF 的链接似乎已损坏。如果您还有那个 PDF，可以分享一下标题吗？

【解决方案3】：

首先，您应该向自己证明您的程序可以在多个内核上运行更快。许多操作系统尽可能努力在同一内核上运行程序线程。

在同一个内核上运行有很多优点。 CPU 缓存是热的，这意味着该程序的数据被加载到 CPU 中。 lock/monitor/synchronization 对象位于 CPU 缓存中，这意味着其他 CPU 不需要跨总线进行缓存同步操作（昂贵！）。

可以很容易地使您的程序始终在同一个 CPU 上运行的一件事是过度使用锁和共享内存。你的线程不应该互相交谈。您的线程在同一内存中使用相同对象的频率越低，它们在不同 CPU 上运行的频率就越高。它们使用相同内存的频率越高，它们必须越频繁地阻塞等待另一个线程。

只要操作系统发现另一个线程的一个线程阻塞，它就会尽可能在同一个 CPU 上运行该线程。它减少了在 CPU 间总线上移动的内存量。我猜这就是导致您在程序中看到的原因。

【讨论】：

【解决方案4】：

我认为这个问题与 Java Parallel Proccesing Framework (JPPF) 有关。使用它，您可以在不同的处理器上运行不同的作业。

【讨论】：

【解决方案5】：

在 Java 中无法设置 CPU 亲和性。 http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4234402

如果必须这样做，请使用 JNI 创建本地线程并设置它们的亲和性。

【讨论】：

【解决方案6】：

最简单的做法是将您的程序分成多个进程。操作系统将在内核之间分配它们。

更难的是把你的程序分成多个线程并信任 JVM 来正确分配它们。这通常是人们利用可用硬件所做的事情。

编辑

多处理程序如何“更简单”？这是管道中的一个步骤。

public class SomeStep {
    public static void main( String args[] ) {
        BufferedReader stdin= new BufferedReader( System.in );
        BufferedWriter stdout= new BufferedWriter( System.out );
        String line= stdin.readLine();
        while( line != null ) {
             // process line, writing to stdout
             line = stdin.readLine();
        }
    }
}

管道中的每个步骤都具有类似的结构。包含任何处理的 9 行开销。

这可能不是绝对最有效的。但这很容易。

并发进程的整体结构不是 JVM 问题。这是操作系统的问题，所以使用 shell。

java -cp pipline.jar FirstStep | java -cp pipline.jar SomeStep | java -cp pipline.jar LastStep

剩下的唯一事情就是为管道中的数据对象进行一些序列化。标准序列化效果很好。阅读http://java.sun.com/developer/technicalArticles/Programming/serialization/ 以获取有关如何序列化的提示。您可以将BufferedReader 和BufferedWriter 替换为ObjectInputStream 和ObjectOutputStream 来完成此操作。

【讨论】：

多进程应用程序如何比多线程应用程序更容易实现？
@S. Lott：我找不到一种简单的方法来使用它，比如说，服务器为每个客户端使用一个进程/线程，并共享可由任何进程/线程修改的数据结构。
不确定多个进程是否一定会有所帮助-取决于您的操作系统，它可能无论如何都在线程级别进行调度。
@Lott：如果你的目标是表现，那对你没有多大好处，不是吗？您基本上是在制作一个较慢版本的消息传递接口。我同意分离处理阶段，但是当您可以使用工作队列和工作线程时，为什么要通过 Stream 这样做呢？
@Lott 同样，仅在 C 语言中快速——问题是 Java 的流 I/O 被同步并在每次 I/O 调用时检查，而不是管道。也不是更容易——如果您使用标准输出/标准输入，您需要定义一个通信协议并可能使用解析。不要忘记写入 StdOut 的异常！使用管理器线程、ExecutorServices 和 Runnable/Callable 任务实现起来要简单得多。它可以在不到 100 行非常简单的代码（带有错误检查）中实现，可能非常快，并且性能良好。

【解决方案7】：

首先，我建议阅读"Concurrency in Practice" by Brian Goetz。

这是迄今为止描述并发 Java 编程的最佳书籍。

并发是“易学难精”。我建议在尝试之前阅读大量有关该主题的内容。让一个多线程程序在 99.9% 的时间内正常工作并在 0.1% 的情况下失败是非常容易的。不过，这里有一些提示可以帮助您入门：

有两种常见的方法可以让程序使用多个内核：

使程序使用多个进程运行。一个例子是使用 Pre-Fork MPM 编译的 Apache，它将请求分配给子进程。在多进程程序中，默认情况下不共享内存。但是，您可以跨进程映射共享内存的部分。 Apache 使用它的“记分板”来做到这一点。
使程序多线程。在多线程程序中，默认情况下所有堆内存都是共享的。每个线程仍然有自己的堆栈，但可以访问堆的任何部分。通常，大多数 Java 程序都是多线程的，而不是多进程的。

在最低级别，可以create and destroy threads。 Java 使得以可移植的跨平台方式创建线程变得容易。

由于创建和销毁线程总是变得昂贵，Java 现在包括 Executors 来创建可重用的线程池。可以将任务分配给执行者，并且可以通过 Future 对象检索结果。

通常，一个任务可以分为更小的任务，但最终结果需要重新组合在一起。例如，使用归并排序，可以将列表分成越来越小的部分，直到每个核心都进行排序。但是，由于每个子列表都已排序，因此需要将其合并才能获得最终的排序列表。由于这是“分而治之”的问题相当普遍，因此有一个JSR framework 可以处理基础分布和加入。这个框架很可能会包含在 Java 7 中。

【讨论】：

JSR 166y 框架已包含在 Java 7 中 java.util.concurrent 包的类 ForkJoinPool 和 ForkJoinTask docs.oracle.com/javase/tutorial/essential/concurrency/…

【解决方案8】：

当我运行它时，它似乎只使用一个 CPU，直到它需要更多使用另一个 CPU - 我有什么可以在Java中强制不同在不同的线程上运行内核/CPU？

我将您问题的这一部分解释为您已经解决了使应用程序支持多线程的问题。尽管如此，它并不会立即开始使用多核。

“有什么办法可以强制……”的答案不是直接的（AFAIK）。您的 JVM 和/或主机操作系统决定使用多少“本机”线程，以及这些线程如何映射到物理处理器。您确实有一些调整选项。例如，我找到了this page，它讨论了如何在 Solaris 上调整 Java 线程。 this page 谈到了其他可能会降低多线程应用程序速度的事情。

【讨论】：

【解决方案9】：

在 Java 中有两种基本的多线程方式。您使用这些方法创建的每个逻辑任务都应在需要且可用时在新内核上运行。

方法一：定义一个Runnable或Thread对象（可以在构造函数中取一个Runnable）并用Thread.start()方法启动它运行。它将在操作系统为其提供的任何内核上执行——通常是负载较少的内核。

教程：Defining and Starting Threads

方法二：定义实现Runnable（如果它们不返回值）或Callable（如果它们返回值）接口的对象，其中包含您的处理代码。将这些作为任务从 java.util.concurrent 包传递给 ExecutorService。 java.util.concurrent.Executors 类有很多方法来创建标准的、有用的 ExecutorServices。 Link 致执行者教程。

根据个人经验，Executors 固定和缓存线程池非常好，尽管您需要调整线程数。 Runtime.getRuntime().availableProcessors() 可用于在运行时计算可用内核数。您需要在应用程序完成后关闭线程池，否则应用程序不会退出，因为 ThreadPool 线程仍在运行。

获得良好的多核性能有时很棘手，而且充满了陷阱：

磁盘 I/O 在运行时会减慢很多平行线。一次只能有一个线程读取/写入磁盘。
对象的同步为多线程操作提供了安全性，但会减慢工作速度。
如果任务太琐碎的（小工作位，执行快速）管理它们的开销在 ExecutorService 中的成本超过您可以从多个内核中获益。
创建新的 Thread 对象很慢。如果可能，ExecutorServices 将尝试重用现有线程。
当多个线程处理某事时，可能会发生各种疯狂的事情。让您的系统保持简单，并尝试使任务在逻辑上不同且不交互。

另一个问题：控制工作很难！一个好的做法是拥有一个创建和提交任务的管理器线程，然后是一对带有工作队列的工作线程（使用 ExecutorService）。

这里我只是谈到了关键点——多线程编程被许多专家认为是最难的编程主题之一。它不直观、复杂，而且抽象通常很弱。

编辑——使用 ExecutorService 的示例：

public class TaskThreader {
    class DoStuff implements Callable {
       Object in;
       public Object call(){
         in = doStep1(in);
         in = doStep2(in);
         in = doStep3(in); 
         return in;
       }
       public DoStuff(Object input){
          in = input;
       }
    }

    public abstract Object doStep1(Object input);    
    public abstract Object doStep2(Object input);    
    public abstract Object doStep3(Object input);    

    public static void main(String[] args) throws Exception {
        ExecutorService exec = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        ArrayList<Callable> tasks = new ArrayList<Callable>();
        for(Object input : inputs){
           tasks.add(new DoStuff(input));
        }
        List<Future> results = exec.invokeAll(tasks);
        exec.shutdown();
        for(Future f : results) {
           write(f.get());
        }
    }
}

【讨论】：

太棒了！我去阅读了有关该主题的更多信息，因为我不清楚 Executors 的优势。我还不确定其他的，但FixedThreadPool 似乎很棒，因为它限制了运行的线程数量（1）避免更改任务的过载，以及（2）确保一些线程首先完成（并快速获得一些结果） .这对于运行实验特别有用。

【解决方案10】：

你应该编写你的程序来完成它的工作，以 lot 的形式将 Callable 交给一个 ExecutorService 并使用 invokeAll(...) 执行。

然后，您可以在运行时从 Executors 类中选择合适的实现。一个建议是调用 Executors.newFixedThreadPool() ，其数字大致对应于 cpu 核心数以保持忙碌。

【讨论】：