在独立文件夹中异步调用CLI工具的实现方案咨询

核心思路 #

你需要创建绑定固定文件夹的长期运行工作进程，每个进程持续监听任务，完成后立即等待下一个任务；同时维护一个调度器，跟踪各工作进程的空闲状态，将新任务动态分配给第一个空闲的工作者。mirai的异步通信机制正好适合实现这种"工作池+动态调度"的模式。

步骤1：定义绑定文件夹的工作者函数 #

每个工作者会常驻在指定文件夹里，循环等待任务，收到任务后执行CLI命令并返回结果。用mirai::daemon创建长期运行的工作进程：

library(mirai)
library(purrr)

# 定义工作者逻辑：绑定文件夹，循环处理任务
folder_worker <- function(folder_path) {
  # 切换到目标文件夹，仅执行一次
  setwd(folder_path)
  message(sprintf("Worker started in folder: %s", folder_path))
  
  # 循环监听任务
  while(TRUE) {
    # 接收调度器发来的任务（CLI命令参数）
    task <- receive()
    if (is.null(task)) break  # 收到终止信号时退出
    
    # 执行CLI命令，wait=TRUE确保完成后再返回
    result <- system(task$cmd, intern = TRUE, wait = TRUE)
    # 将结果发回调度器，同时标记自身空闲
    send(list(worker_folder = folder_path, result = result, status = "done"))
  }
}

步骤2：初始化工作池 #

根据你的文件夹列表，创建对应数量的工作者进程，每个绑定到指定文件夹，同时记录每个工作者的mirai对象和状态：

# 你的目标文件夹列表
folders <- c("folder1", "folder2", "folder3")

# 初始化工作池：创建工作者并跟踪状态
worker_pool <- map(folders, function(folder) {
  # 创建daemon工作进程，传入文件夹路径
  m <- mirai(folder_worker(folder_path = .x), .args = list(folder))
  list(mirai = m, folder = folder, is_idle = TRUE)
})

步骤3：实现动态任务调度逻辑 #

维护任务队列，轮询空闲工作者分配任务；同时监听工作者的完成信号，标记其为空闲并处理结果，再分配下一个任务：

# 示例任务列表：每个任务是要执行的CLI命令
tasks <- c(
  "tool1 --input data1.txt",
  "tool1 --input data2.txt",
  "tool1 --input data3.txt",
  "tool1 --input data4.txt"
)

# 任务索引，用于遍历任务队列
task_idx <- 1
# 存储所有任务结果
task_results <- list()

# 调度循环：直到所有任务完成
while(task_idx <= length(tasks) || any(map_lgl(worker_pool, ~!$.x$is_idle))) {
  # 找到第一个空闲的工作者
  idle_worker <- detect(worker_pool, ~$.x$is_idle)
  
  # 如果有空闲工作者且还有未分配任务
  if (!is.null(idle_worker) && task_idx <= length(tasks)) {
    # 发送任务给空闲工作者
    send(idle_worker$mirai, list(cmd = tasks[task_idx]))
    # 标记工作者为忙碌
    worker_pool[[which(map_chr(worker_pool, ~$.x$folder) == idle_worker$folder)]]$is_idle <- FALSE
    task_idx <- task_idx + 1
  }
  
  # 检查是否有工作者完成任务
  completed_workers <- keep(worker_pool, ~!$.x$is_idle && resolved(.x$mirai))
  walk(completed_workers, function(w) {
    # 获取任务结果
    result <- w$mirai$data
    task_results[[length(task_results)+1]] <- result
    message(sprintf("Worker in %s completed task: %s", result$worker_folder, paste(result$result, collapse = " ")))
    
    # 标记工作者为空闲
    worker_pool[[which(map_chr(worker_pool, ~$.x$folder) == w$folder)]]$is_idle <- TRUE
  })
  
  # 短暂休眠，避免轮询过于频繁
  Sys.sleep(0.1)
}

# 所有任务完成后，终止工作者进程
walk(worker_pool, ~send(.x$mirai, NULL))

结合purrr::map的简化方案（开发版purrr） #

如果想用开发版purrr的异步map能力，可以把工作池逻辑封装成自定义映射函数，核心逻辑不变：

# 自定义异步map函数，绑定文件夹工作者
folder_map <- function(tasks, folders) {
  # 初始化工作池
  worker_pool <- map(folders, function(folder) {
    m <- mirai(folder_worker(folder_path = .x), .args = list(folder))
    list(mirai = m, folder = folder, is_idle = TRUE)
  })
  
  task_idx <- 1
  results <- list()
  
  # 调度循环
  while(task_idx <= length(tasks) || any(map_lgl(worker_pool, ~!$.x$is_idle))) {
    idle_worker <- detect(worker_pool, ~$.x$is_idle)
    if (!is.null(idle_worker) && task_idx <= length(tasks)) {
      send(idle_worker$mirai, list(cmd = tasks[task_idx]))
      worker_pool[[which(map_chr(worker_pool, ~$.x$folder) == idle_worker$folder)]]$is_idle <- FALSE
      task_idx <- task_idx + 1
    }
    
    completed_workers <- keep(worker_pool, ~!$.x$is_idle && resolved(.x$mirai))
    walk(completed_workers, function(w) {
      result <- w$mirai$data
      results[[length(results)+1]] <- result
      worker_pool[[which(map_chr(worker_pool, ~$.x$folder) == w$folder)]]$is_idle <- TRUE
    })
    Sys.sleep(0.1)
  }
  
  # 终止工作者
  walk(worker_pool, ~send(.x$mirai, NULL))
  return(results)
}

# 使用示例
task_results <- folder_map(tasks, folders)

关键注意事项 #

• 工作者常驻性：mirai::daemon创建的进程会长期运行，直到收到send(NULL)终止信号，避免了反复创建进程和切换文件夹的开销。
• 状态准确性：必须严格跟踪工作者的空闲/忙碌状态，防止重复分配任务。
• 错误处理：可以在folder_worker中添加tryCatch块，捕获CLI命令执行的错误，将错误信息返回调度器，避免工作者崩溃。
• CLI参数适配：根据实际CLI工具的需求，调整system()函数的参数（比如intern=FALSE直接输出到控制台）。

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