0Pricing
R Academy · 课时

future 框架

使用 plan(multisession) 和 future() 实现异步、可移植的并行处理

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

什么是 future 包

future 包为 R 中的异步和并行编程提供了统一的高级 API。future 是一个值的占位符，该值稍后才会生成——可能是在另一个核心或另一台计算机上计算得到。

library(future)

# A simple future: computation happens asynchronously
f <- future({
  Sys.sleep(0.5)  # simulate slow work
  42
})

cat('Future created, doing other work...\n')

# Retrieve the result (blocks until done)
result <- value(f)
cat('Result:', result, '\n')

plan()：选择后端

plan() 为之后创建的所有 future 设置执行策略。最常用的策略包括 sequential（默认，单线程）、multisession（多个 R 会话，适用于所有平台）和 multicore（通过派生进程运行，仅适用于 Unix/macOS）。

library(future)

# Default: sequential (no parallelism)
plan(sequential)
cat('Strategy:', class(plan())[1], '\n')

# Parallel with separate R sessions (works on Windows too)
plan(multisession, workers = 4)
cat('Strategy:', class(plan())[1], '\n')

# Forking (Unix/macOS only, lower overhead)
if (.Platform$OS.type != 'windows') {
  plan(multicore, workers = 4)
  cat('Strategy:', class(plan())[1], '\n')
}

# Reset to sequential
plan(sequential)

value()：等待结果

value(f) 会阻塞当前进程，直到 future 的计算完成，然后返回结果。如果 future 抛出了错误，value() 会在调用它的会话中再次抛出该错误。

library(future)
plan(multisession, workers = 2)

# Launch two slow tasks concurrently
f1 <- future({ Sys.sleep(0.3); sum(1:1000) })
f2 <- future({ Sys.sleep(0.3); prod(1:10) })

# Both run in parallel; total time ~ 0.3s not 0.6s
start <- proc.time()[['elapsed']]
v1 <- value(f1)
v2 <- value(f2)
elapsed <- proc.time()[['elapsed']] - start

cat('v1:', v1, ' v2:', v2, '\n')
cat('Elapsed:', round(elapsed, 2), 's\n')

plan(sequential)

%<-% 运算符

%<-% 运算符是 future() 加延迟执行的 value() 的语法糖。它看起来像普通赋值，但右侧会异步运行。第一次访问该变量时，才会获取对应的值。

library(future)
plan(multisession, workers = 2)

# %<-% starts the computation immediately in the background
x %<-% {
  Sys.sleep(0.3)
  rnorm(5, mean = 10)
}

y %<-% {
  Sys.sleep(0.3)
  runif(5, min = 1, max = 5)
}

cat('Both running in background...\n')

# Accessing x or y here blocks until ready
cat('x:', round(x, 2), '\n')
cat('y:', round(y, 2), '\n')

plan(sequential)

自动检测全局变量

future 最强大的功能之一是全局变量检测：它会自动找出 future 代码在调用环境中引用的变量，并将这些变量发送给工作进程——无需手动调用 clusterExport()。

library(future)
plan(multisession, workers = 2)

# These globals are auto-detected and sent to the worker
threshold <- 50
multiplier <- 3

f <- future({
  x <- threshold * multiplier  # uses both globals
  x + 1
})

cat('Result:', value(f), '\n')  # 151

# Inspect which globals were identified
future_obj <- future({
  threshold + multiplier
}, lazy = TRUE)
cat('Globals found:', paste(names(future::getGlobalsAndPackages(future_obj)$globals), collapse = ', '), '\n')

plan(sequential)

future_options：调整行为

future.options（通过 options() 或 future::plan() 的参数访问）用于控制全局设置，包括每个 future 允许使用的最大内存、全局变量检测策略以及超时行为。

library(future)

# Cap memory per future at 500 MB
options(future.globals.maxSize = 500 * 1024^2)  # bytes

# Disable automatic global detection (manual control)
options(future.globals.onReference = 'error')

# Inspect current settings
cat('Max globals size (MB):',
    getOption('future.globals.maxSize') / 1024^2, '\n')

# Restore defaults
options(future.globals.maxSize = 500 * 1024^2)
options(future.globals.onReference = NULL)

plan(multisession, workers = 2)

# Large object: you'd get an error if it exceeds the cap
big_vec <- 1:1e5  # small enough
f <- future(sum(big_vec))
cat('Sum:', value(f), '\n')

plan(sequential)

嵌套 Future

Future 可以嵌套：一个 future 本身可以创建内部 future。默认情况下，内部 future 会按顺序运行，除非您使用 plan(list(...)) 设置嵌套计划。这对于两层并行很有用。

library(future)

# Two-level parallelism: outer uses multisession, inner sequential
plan(list(multisession, sequential))

outer_futures <- lapply(1:3, function(i) {
  future({
    # Each outer worker runs its own sequential inner work
    inner <- lapply(1:4, function(j) i * j)
    unlist(inner)
  })
})

results <- lapply(outer_futures, value)
for (i in seq_along(results)) {
  cat('Outer', i, ':', results[[i]], '\n')
}

plan(sequential)

使用 Future 处理错误

当 future 的计算抛出错误时，value() 会再次抛出该错误。将 value() 放在 tryCatch() 中，可以处理单个 future 的错误，而不会停止整个处理流程。

library(future)
plan(multisession, workers = 2)

# This future will error
bad_future <- future(log(-1, base = 'oops'))  # invalid arg
good_future <- future(sqrt(144))

# Handle errors gracefully
bad_result <- tryCatch(
  value(bad_future),
  error = function(e) {
    cat('Caught error:', conditionMessage(e), '\n')
    NA
  }
)

good_result <- value(good_future)
cat('bad_result:', bad_result, '\n')
cat('good_result:', good_result, '\n')

plan(sequential)

检查 Future 状态

如果某个 future 已完成，resolved(f) 会在不阻塞的情况下返回 TRUE。这样，您就可以轮询完成状态，并在等待期间执行有用的工作，从而创建非阻塞的轮询循环。

library(future)
plan(multisession, workers = 2)

f <- future({
  Sys.sleep(0.5)
  'done'
})

# Poll without blocking
counter <- 0
while (!resolved(f)) {
  counter <- counter + 1
  Sys.sleep(0.1)
  cat('Polling... (', counter, ')\n')
}

cat('Future resolved! Result:', value(f), '\n')
cat('Polled', counter, 'times\n')

plan(sequential)

将 future 与 lapply 结合

将 lapply() 与 future() 结合，可以分发 N 个任务；再将 lapply() 与 value() 结合，可以收集结果。这种模式让您能够明确控制每个 future 的启动时间。

library(future)
plan(multisession, workers = 4)

# Fan out: create all futures
params <- list(
  list(n = 1000, mean = 0),
  list(n = 1000, mean = 5),
  list(n = 1000, mean = 10),
  list(n = 1000, mean = -3)
)

futures <- lapply(params, function(p) {
  future({
    x <- rnorm(p$n, mean = p$mean)
    list(mean = mean(x), sd = sd(x))
  })
})

# Fan in: collect results
results <- lapply(futures, value)
for (i in seq_along(results)) {
  cat('Param mean', params[[i]]$mean,
      '-> observed mean:', round(results[[i]]$mean, 3), '\n')
}

plan(sequential)

实际案例：评估交叉验证折

交叉验证是一个非常适合并行处理的问题：每个折彼此独立。使用 future，您可以同时评估全部 K 个折，从而显著缩短计算复杂模型所需的实际耗时。

library(future)
plan(multisession, workers = 4)

set.seed(42)
data <- data.frame(x = rnorm(200), y = rnorm(200))
fold_ids <- sample(rep(1:4, 50))

# Launch all folds in parallel
fold_futures <- lapply(1:4, function(k) {
  future({
    train <- data[fold_ids != k, ]
    test  <- data[fold_ids == k, ]
    fit   <- lm(y ~ x, data = train)
    preds <- predict(fit, newdata = test)
    rmse  <- sqrt(mean((test$y - preds)^2))
    rmse
  })
})

rmses <- unlist(lapply(fold_futures, value))
cat('Fold RMSEs:', round(rmses, 4), '\n')
cat('Mean RMSE:', round(mean(rmses), 4), '\n')

plan(sequential)

快速检查

您希望运行一段同时适用于 Windows 和 macOS 的并行代码。应该选择哪种 plan() 策略？

回顾：future 框架

要点：

  • future({expr}) 异步运行一个表达式；value(f) 获取结果
  • %<-% 是语法糖，表示带延迟 future 求值的赋值
  • 跨平台运行使用 plan(multisession)；在 Unix/macOS 上使用 plan(multicore) 通过派生进程运行
  • 全局变量会被自动检测并发送给工作进程——无需调用 clusterExport()
  • resolved(f) 可以在不阻塞的情况下检查是否完成
  • 将 value() 放在 tryCatch() 中，可以构建能够从错误中恢复的处理流程
  • 完成后始终调用 plan(sequential) 或 plan() 进行重置
library(future)
plan(multisession, workers = 2)

# Concise pattern: fan-out then fan-in
task <- function(i) future({ i^2 + i })
results <- lapply(1:6, task)
cat(unlist(lapply(results, value)), '\n')  # 2 6 12 20 30 42

plan(sequential)

常见问题解答

「future 框架」课时是免费的吗？

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

「future 框架」这节课中我会学到什么？

使用 plan(multisession) 和 future() 实现异步、可移植的并行处理 你通过在浏览器中直接运行的动手代码来练习 R Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 R Academy 需要有经验吗？

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

「future 框架」课时需要多长时间？

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

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

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

此课程中的所有课时

  1. parallel 软件包与 detectCores()
  2. future 框架
  3. furrr：并行执行 purrr 操作
  4. 调试并行代码与负载均衡
← 返回 R Academy