我应该如何为并行计算生成线程？答案

【问题标题】：How should I spawn threads for parallel computation?我应该如何为并行计算生成线程？
【发布时间】：2015-06-29 00:08:01
【问题描述】：

今天，我进入了多线程。由于这是一个新概念，我想我可以通过将一个简单的迭代转化为一个并行化的迭代来开始学习。但是，我想我还没开始就卡住了。

最初，我的循环看起来像这样：

let stuff: Vec<u8> = items.into_iter().map(|item| {
    some_item_worker(&item)
}).collect();

我在items 中放入了相当多的东西，完成计算大约需要 0.05 秒。所以，当我成功实现多线程后，看到时间减少了，我真的很兴奋！

当我使用线程时，我遇到了麻烦，可能是因为我的推理错误。

use std::thread;

let threads: Vec<_> = items.into_iter().map(|item| {
    thread::spawn(move || {
        some_item_worker(&item)
    })
}).collect(); // yeah, this is followed by another iter() that unwraps the values

我有一个四核 CPU，这意味着我最多只能同时运行 4 个线程。我猜它是这样工作的：一旦迭代器启动，就会产生线程。每当一个线程结束时，另一个线程就会开始，因此在任何给定时间，4 个线程同时运行。

结果是（在重新运行后）大约 0.2 秒才能完成相同的计算。显然，这里没有进行并行计算。我不知道为什么时间增加了4倍，但我确定我误解了一些东西。

既然这不是正确的方法，我应该如何修改程序以使线程同时执行？

编辑：

对不起，我错了大约 0.2 秒。当我注意到通常的迭代运行了 2 秒时，我醒来并再次尝试。事实证明，某些进程一直在疯狂地消耗内存。当我重新启动系统并再次尝试线程迭代时，它运行了大约 0.07 秒。以下是每次运行的一些时间安排。

实际迭代（第一次）：

0.0553760528564 seconds
0.0539519786835 seconds
0.0564560890198 seconds

线程一：

0.0734670162201 seconds
0.0727820396423 seconds
0.0719120502472 seconds

我同意线程确实在同时运行，但完成这项工作似乎又需要 20 毫秒。我的实际目标是利用我的处理器并行运行线程并尽快完成工作。这会很复杂吗？我应该怎么做才能使这些线程并行运行，而不是并发运行？

【问题讨论】：

如果程序受内存限制而不是 CPU 限制，那么多线程将无济于事，并且可能由于缓存冲突而减慢速度。多线程对于 CPU 密集型计算很有用，或者作为异步 I/O 的替代方法来分离设备（其中每个线程执行常规 I/O 并等待设备）。
实际上很不清楚你在问什么。您的标题是提到同步，但随后您的身体会询问如何让事情同时发生。线程间同步是一个很大的话题，但您似乎并没有问这个问题。
@Shepmaster 是的，这令人困惑。现在修复:)
有多少项，即每个some_item_worker 需要多长时间？产生一个线程并等待它完成需要时间。如果每个线程只做非常少量的工作，那么生成大量线程很容易降低性能。
这并不一定意味着您根本无法进行并行化，您只需要生成更少的线程并让每个线程有更多的工作要做。

标签： multithreading concurrency rust

【解决方案1】：

我有一个四核 CPU，这意味着我最多只能同时运行 4 个线程。

只有 4 个可能同时运行，但您当然可以创建超过 4 个...

每当一个线程结束时，另一个线程就会开始，因此在任何给定时间，4 个线程同时运行（这只是一个猜测）。

每当您有一个猜测时，您都应该创建一个实验来确定您的猜测是否正确。这是一个：

use std::{iter, thread, time::Duration};

fn main() {
    let items: Vec<_> = iter::repeat(0).take(500).collect();

    let threads: Vec<_> = items
        .into_iter()
        .map(|_| {
            thread::spawn(move || {
                println!("Started!");
                thread::sleep(Duration::from_millis(500));
                println!("Finished!");
            })
        })
        .collect();

    for handle in threads {
        handle.join().unwrap()
    }
}

如果你运行它，你会看到“Started!”打印出 500 次，然后是 500 个“Finished!”

显然，这里没有进行并行计算

很遗憾，您的问题不够详细，我们无法说明为什么您的时间会增加。在我提供的示例中，它花费的时间不到 600 毫秒，所以它显然不是串行发生的！

【讨论】：

我现在明白线程是同时运行的。另外，如果不清楚，我很抱歉（我现在已经改进了这个问题）。运行程序时，一些糟糕的进程一直在消耗我的内存。重新启动后，问题得到解决，但我仍然担心时间问题。我的意思是，线程迭代似乎需要额外的 200 毫秒。你能解释一下吗？ :)
读了一些之后，我才意识到并发和并行是两个独立的事物。而且，就我而言，我需要的是并行化。那么，并行运行这些线程会很复杂吗？如果是这样，你可以给我一些提示，让我开始。我在研究周围没有任何问题:)

【解决方案2】：

创建线程是有代价的。如果线程内部的计算成本足够小，那么与线程成本或线程造成的低效率相比，它就会相形见绌。

例如，生成 1000 万个线程以使 1000 万个 u8 增加一倍可能并不值得。对其进行矢量化可能会产生更好的性能。

也就是说，您仍然可以通过并行化廉价任务来获得一些改进。但是您想通过线程池使用更少的线程，而线程数量很少（因此您在任何给定点创建的线程数量（少量），CPU 争用更少）或更复杂的东西（在引擎盖下，api很简单）就像Rayon。

// Notice `.par_iter()` turns it into a `parallel iterator`
let stuff: Vec<u8> = items.par_iter().map(|item| {
    some_item_worker(&item)
}).collect();

【讨论】：