R Academy · 课时

parallel 软件包与 detectCores()

Q: 「parallel 软件包与 detectCores()」这节课中我会学到什么？

启动分叉式或套接字集群，并在 CPU 核心之间分配工作 你通过在浏览器中直接运行的动手代码来练习 R Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

启动分叉式或套接字集群，并在 CPU 核心之间分配工作

第 1 / 4 课13 个步骤

parallel 软件包与 detectCores() 是 CoddyKit 上的免费 R Academy 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 R Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 R Academy 课程共包含 4 节课。

为什么需要并行计算

现代计算机拥有多个 CPU 核心。默认情况下，R 只使用一个核心，其他核心处于空闲状态。R 内置的 parallel 包可以让您利用所有核心，加快重复计算。

# Check how many cores your machine has
library(parallel)

total_cores <- detectCores()
logical_cores <- detectCores(logical = TRUE)
physical_cores <- detectCores(logical = FALSE)

cat('Total logical cores:', total_cores, '
')
cat('Physical cores:', physical_cores, '
')

makeCluster 和 stopCluster

makeCluster(n) 会创建 n 个工作进程。完成后始终调用 stopCluster(cl) 以释放资源。常见做法是使用 detectCores() - 1，为 OS 留出一个核心。

library(parallel)

# Spawn workers (leave 1 core for system)
n_cores <- detectCores() - 1
cl <- makeCluster(n_cores)

cat('Cluster created with', n_cores, 'workers\n')

# Always clean up!
stopCluster(cl)
cat('Cluster stopped.\n')

clusterExport：共享变量

工作进程拥有各自的内存空间，无法看到您的全局环境。使用 clusterExport(cl, varlist) 将指定名称的对象从主进程复制到所有工作进程。

library(parallel)
cl <- makeCluster(2)

# Define a variable and a function in the master
base_value <- 100
add_base <- function(x) x + base_value

# Export them to workers
clusterExport(cl, varlist = c('base_value', 'add_base'))

# Now workers can use them
result <- parLapply(cl, 1:4, function(x) add_base(x))
cat(unlist(result), '\n')  # 101 102 103 104

stopCluster(cl)

clusterEvalQ：运行设置代码

clusterEvalQ(cl, expr) 会在每个工作进程上计算一个表达式。在主要计算开始前，它适合用于在所有节点上加载包或加载辅助文件。

library(parallel)
cl <- makeCluster(2)

# Load a package on every worker
clusterEvalQ(cl, {
  library(stats)
  set.seed(42)  # set seed per worker
})

# Each worker can now use stats functions
result <- parLapply(cl, 1:4, function(n) rnorm(n, mean = 0, sd = 1))
result[[1]]  # one random normal value

stopCluster(cl)

parLapply：并行 lapply

parLapply(cl, X, FUN) 是 lapply() 的并行等价函数。它会将 X 的元素分配给各个工作进程，并将结果收集为列表。它适用于所有平台（Windows、macOS、Linux）。

library(parallel)
cl <- makeCluster(2)

# Simulate slow computation: sleep 0.1s per item
clusterEvalQ(cl, Sys.sleep)

slow_square <- function(x) {
  Sys.sleep(0.05)
  x^2
}

clusterExport(cl, 'slow_square')

system.time(
  result <- parLapply(cl, 1:8, slow_square)
)
cat(unlist(result), '\n')  # 1 4 9 16 25 36 49 64

stopCluster(cl)

parSapply：简化结果

parSapply(cl, X, FUN) 是 sapply() 的并行版本。在可能的情况下，它会自动将结果简化为向量或矩阵；当您的函数返回标量时，这会非常方便。

library(parallel)
cl <- makeCluster(2)

# parSapply returns a simplified vector
squares <- parSapply(cl, 1:10, function(x) x^2)
cat(squares, '\n')  # 1 4 9 16 25 36 49 64 81 100

# Returns a matrix if FUN returns a vector of same length
stats_result <- parSapply(cl, 1:4, function(n) {
  x <- rnorm(100)
  c(mean = mean(x), sd = sd(x))
})
print(stats_result)  # 2x4 matrix

stopCluster(cl)

mclapply：基于派生的并行处理

mclapply() 使用派生机制——它会复制当前 R 进程，而不是创建新进程，因此启动速度更快，工作进程也能隐式访问父进程环境。不过，它只能在 Unix/macOS 上运行，不能在 Windows 上运行。

library(parallel)

# mclapply: Unix/macOS only
# Workers inherit the parent environment automatically
base_value <- 42

if (.Platform$OS.type != 'windows') {
  result <- mclapply(
    1:8,
    function(x) x * base_value,  # base_value visible without export
    mc.cores = 4
  )
  cat(unlist(result), '\n')
} else {
  cat('mclapply not supported on Windows. Use parLapply instead.\n')
}

对比顺序处理与并行处理的性能

并行处理存在开销：启动集群、序列化数据以及进程间通信都需要时间。只有当每个元素的计算成本足够高，能够抵消这些开销时，并行处理才会带来收益。

library(parallel)
cl <- makeCluster(2)

# Task: compute 100 iterations of matrix multiply
heavy_task <- function(n) {
  m <- matrix(rnorm(200), nrow = 100)
  sum(m %*% t(m))
}
clusterExport(cl, 'heavy_task')

seq_time <- system.time(lapply(1:20, heavy_task))[['elapsed']]
par_time <- system.time(parLapply(cl, 1:20, heavy_task))[['elapsed']]

cat('Sequential:', round(seq_time, 3), 's\n')
cat('Parallel:  ', round(par_time, 3), 's\n')
cat('Speedup:   ', round(seq_time / par_time, 2), 'x\n')

stopCluster(cl)

在并行处理中设置 RNG 种子

并行生成随机数比较棘手——每个工作进程都需要独立且可重复的随机数流。使用 clusterSetRNGStream(cl, seed) 和 L'Ecuyer-CMRG 生成器，即可获得可重复的并行随机结果。

library(parallel)
cl <- makeCluster(2)

# Set reproducible RNG streams across workers
clusterSetRNGStream(cl, iseed = 123)

# Each worker uses its own independent random stream
results1 <- parSapply(cl, 1:6, function(i) rnorm(1))

# Reset and repeat — same results
clusterSetRNGStream(cl, iseed = 123)
results2 <- parSapply(cl, 1:6, function(i) rnorm(1))

cat('Run 1:', round(results1, 4), '\n')
cat('Run 2:', round(results2, 4), '\n')
cat('Identical:', identical(results1, results2), '\n')

stopCluster(cl)

集群中的错误处理

如果某个工作进程抛出错误，parLapply() 会在主进程中重新抛出该错误。您可以在函数内部使用 tryCatch() 包装调用，也可以在整个 parLapply() 调用外部使用 tryCatch()，以便优雅地处理失败。

library(parallel)
cl <- makeCluster(2)

# Wrap risky code inside the worker function
safe_log <- function(x) {
  tryCatch(
    log(x),
    warning = function(w) NA_real_,
    error   = function(e) NA_real_
  )
}
clusterExport(cl, 'safe_log')

# -1 produces NaN warning, 'a' produces an error
input <- list(4, 9, -1, 'a', 16)
result <- parLapply(cl, input, safe_log)
cat(unlist(result), '\n')  # 1.386 2.197 NaN NA 2.773

stopCluster(cl)

实用的并行处理模式

下面是一个完整且符合惯用写法的并行工作流程：检测核心数、创建集群、导出依赖项、执行计算、收集结果，并使用 on.exit() 确保始终停止集群，即使发生错误也是如此。

library(parallel)

run_parallel <- function(data, fn, n_workers = detectCores() - 1) {
  cl <- makeCluster(n_workers)
  on.exit(stopCluster(cl))  # guaranteed cleanup

  clusterExport(cl, 'fn', envir = environment())

  result <- parLapply(cl, data, fn)
  result
}

# Example: bootstrap mean estimation
samples <- lapply(1:100, function(i) rnorm(50, mean = 5, sd = 2))
means <- run_parallel(samples, mean)

cat('Grand mean:', round(mean(unlist(means)), 3), '\n')
cat('95% CI: [',
    round(quantile(unlist(means), 0.025), 3), ',',
    round(quantile(unlist(means), 0.975), 3), ']\n')

快速检查

应使用哪个函数来确保即使并行代码中发生错误，也会调用 stopCluster(cl)？

在交互式课程中回答

回顾：parallel 包

要点：

detectCores() 报告可用核心数；使用 detectCores() - 1 设置集群大小
makeCluster(n) / stopCluster(cl) 管理工作进程的生命周期
clusterExport() 复制对象；clusterEvalQ() 在工作进程上运行设置代码
parLapply() / parSapply() 是跨平台的并行迭代函数
mclapply() 仅支持 Unix，但由于使用派生机制，启动速度更快
使用 on.exit(stopCluster(cl)) 确保完成清理
并行处理只有在计算量很大的任务中才能带来收益

# Minimal reproducible parallel pattern
library(parallel)

cl <- makeCluster(max(1, detectCores() - 1))
on.exit(stopCluster(cl))

clusterSetRNGStream(cl, iseed = 42)
result <- parSapply(cl, 1:8, function(x) x^2 + rnorm(1, 0, 0.1))
cat(round(result, 2), '\n')

免费开始

用 AI 导师学习 R — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 43
课程: 159

免费开始学习查看完整课程

常见问题解答

「parallel 软件包与 detectCores()」课时是免费的吗？

是的 — 「parallel 软件包与 detectCores()」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 R Academy 课程的其余内容，请升级到 CoddyKit PRO。 R Academy 课程共包含 4 节课。

「parallel 软件包与 detectCores()」这节课中我会学到什么？

启动分叉式或套接字集群，并在 CPU 核心之间分配工作你通过在浏览器中直接运行的动手代码来练习 R Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 R Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 R Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「parallel 软件包与 detectCores()」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 R Academy 课中编写并运行代码吗？

能。每节 R Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 R Academy