如何在furrr包的并行进程间实现共享计数器递增？

问题根源 #

在multisession并行模式下，每个worker进程拥有独立的内存空间，你手动维护的全局计数器n无法在进程间共享。每个worker进程的n都会从0开始递增，8核CPU下最多只能显示每个worker完成1个任务后的8/64，无法反映全局的完成进度。

解决方案 #

progressr包本身已针对并行任务的进度跟踪做了优化，无需手动维护计数器。它会自动汇总所有worker进程的进度更新，正确统计全局的已完成任务数。

修改后的代码如下：

library(furrr)
library(progressr)

# 自定义进度条显示格式，包含已完成数/总任务数
handlers(handler_progress(format = "已完成 :current/:total 个任务"))

long_process <- function(items) {
  nb_of_cores <- parallel::detectCores()
  plan(multisession, workers = nb_of_cores)
  
  N <- length(items)
  # 创建进度跟踪器，指定总任务数
  p <- progressr::progressor(along = items)
  
  item_process <- function(item) {
    # 每个任务完成时调用p()，progressr自动累计全局进度
    p()
    Sys.sleep(1)
  }
  
  future_walk(items, item_process)
}

with_progress(long_process(1:64))

关键说明 #

• 移除了手动维护的计数器n，避免进程间的变量隔离问题。
• 通过progressor(along = items)指定总任务数，progressr会自动跟踪全局完成进度。
• 自定义handler_progress的format参数，让进度条直接显示已完成 x/64 个任务的格式，无需在任务函数中拼接消息。

如果需要在任务函数中传递自定义消息（比如当前处理的item内容），可以修改p()的调用为：

p(sprintf("处理任务 %d", item))

此时进度条会显示已完成 x/64 个任务 处理任务 xx的组合信息。

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