如何在parLapply调用中设置.GlobalEnv变量查看任务进度？

这个问题我之前也碰到过！核心原因是并行集群里的每个worker进程都有自己独立的运行环境，你在worker里修改的.GlobalEnv其实是worker自己的全局环境，和主进程的.GlobalEnv完全不是同一个空间，所以主进程根本看不到这个更新。下面给你两个可行的解决方案：

方案一：手动实现进程间通信（无需额外包） #

我们可以放弃parLapply的封装，手动处理任务分发和进度反馈。让每个worker完成任务后，通过sendMaster()给主进程发送进度消息，主进程接收后更新本地的进度变量。

library(parallel)

# 创建集群
cl <- makeCluster(1)
total_tasks <- 3
progress <- 0

# 定义worker执行的函数
task_fun <- function(x) {
  # 模拟任务执行
  Sys.sleep(1)
  # 给主进程发送进度更新信号
  sendMaster(list(type = "progress", value = 1))
  # 返回任务结果
  return(paste("完成任务", x))
}

# 手动分发任务并处理进度
results <- list()
for (i in 1:total_tasks) {
  # 发送任务到worker
  clusterSendCall(cl, task_fun, args = list(i))
  # 接收worker的消息
  msg <- recvMaster()
  if (msg$type == "progress") {
    progress <- progress + msg$value
    cat(sprintf("当前进度：%d/%d\n", progress, total_tasks))
  } else {
    results[[i]] <- msg$value
  }
}

stopCluster(cl)
print(results)

方案二：使用专业进度跟踪包（progressr） #

如果觉得手动通信太麻烦，推荐使用progressr包，它专门支持并行环境下的进度跟踪，用法更简洁。

library(parallel)
library(progressr)

# 启用进度跟踪
handlers(global = TRUE)

cl <- makeCluster(1)
# 导出进度相关的函数到worker
clusterExport(cl, c("p"))

total_tasks <- 3
with_progress({
  p <- progressor(along = 1:total_tasks)
  voos <- parLapply(cl, 1:total_tasks, function(x) {
    # 模拟任务执行
    Sys.sleep(1)
    # 更新进度
    p()
    return(paste("完成任务", x))
  })
})

stopCluster(cl)
print(voos)

这个方案会在控制台自动显示进度条，体验更好，而且不需要手动处理进程间的消息传递。

内容的提问来源于stack exchange，提问作者Daniel