如何修改Shiny+callr代码实现多后台任务并行运行？

问题背景 #

我正在学习R中的异步编程，开发一个Shiny应用：用户通过下拉框创建“场景”生成表格，点击表格行触发一个休眠任务（时长由用户输入），任务完成后该行状态改为“完成”。但当前代码存在问题：启动一个任务后再启动另一个，第一个任务会被终止，需要修改实现多任务后台并行运行。

原代码核心问题 #

• 用reactiveVal(NULL)存储后台进程，新任务启动时会覆盖旧进程对象，导致无法跟踪之前的任务
• 单个row_index和check_finished只能监控一个任务的状态，无法同时处理多个并行任务

修改后的完整代码 #

library(shiny)
library(DT)
library(callr)

# Define server logic
server <- function(input, output, session) {
  
  run_task <- function(sleep_duration) {
    Sys.sleep(sleep_duration)
  }
  
  # 存储表格数据和多后台进程
  data <- reactiveValues(table_data = NULL)
  bg_procs <- reactiveValues()  # 键为行索引，值为对应后台进程对象
  
  observeEvent(input$add, {
    new_data <- data.frame(
      Dropdown1 = input$dropdown1,
      Dropdown2 = input$dropdown2,
      Dropdown3 = input$dropdown3,
      Status = "Pending",
      stringsAsFactors = FALSE
    )
    if (is.null(data$table_data)) {
      data$table_data <- new_data
    } else {
      data$table_data <- rbind(data$table_data, new_data)
    }
  })
  
  observeEvent(input$action, {
    selected_rows <- input$output_table_rows_selected
    if (length(selected_rows) == 0) {
      return()
    }
    
    # 为每个选中行启动独立后台任务
    for (row_idx in selected_rows) {
      # 跳过已运行或完成的任务
      if (data$table_data$Status[row_idx] != "Pending") {
        next
      }
      
      sleep_duration <- as.numeric(data$table_data$Dropdown2[row_idx])
      p <- r_bg(
        func = function(sleep_duration) {
          Sys.sleep(sleep_duration)
        },
        supervise = TRUE,
        args = list(sleep_duration = sleep_duration)
      )
      
      # 记录进程并标记状态为运行中
      bg_procs[[as.character(row_idx)]] <- p
      data$table_data$Status[row_idx] <- "In Progress"
      cat(paste0("\n任务行", row_idx, "启动于", Sys.time(), "\n"))
    }
  })
  
  # 每秒检查所有后台任务状态
  observe({
    invalidateLater(1000)
    
    proc_keys <- names(reactiveValuesToList(bg_procs))
    if (length(proc_keys) == 0) {
      return()
    }
    
    for (key in proc_keys) {
      row_idx <- as.integer(key)
      p <- bg_procs[[key]]
      
      if (!p$is_alive()) {
        # 任务完成，更新状态并移除进程
        data$table_data$Status[row_idx] <- "Completed"
        rm(list = key, envir = bg_procs)
        cat(paste0("\n任务行", row_idx, "完成于", Sys.time(), "\n"))
      }
    }
  })
  
  observeEvent(input$deleteRows, {
    if (!is.null(input$output_table_rows_selected)) {
      selected_rows <- as.numeric(input$output_table_rows_selected)
      # 删除对应后台进程
      for (row_idx in selected_rows) {
        key <- as.character(row_idx)
        if (!is.null(bg_procs[[key]])) {
          bg_procs[[key]]$kill()
          rm(list = key, envir = bg_procs)
        }
      }
      # 删除表格行
      data$table_data <- data$table_data[-selected_rows, ]
    }
  })
  
  # 渲染表格（支持多选）
  output$output_table <- renderDT({
    req(data$table_data)
    datatable(
      data$table_data,
      selection = "multiple",
      options = list(pageLength = 10)
    )
  })
}

# Define UI
ui <- fluidPage(
  titlePanel("Dashboard"),
  
  sidebarLayout(
    sidebarPanel(
      selectInput("dropdown1", "Dropdown 1", choices = c("Option 1", "Option 2", "Option 3")),
      selectInput("dropdown2", "休眠时长", choices = c("1", "10", "60")),
      selectInput("dropdown3", "Dropdown 3", choices = c("Option X", "Option Y", "Option Z")),
      
      actionButton("add", "添加场景"),
      actionButton("edit", "编辑"),
      actionButton("deleteRows", "删除选中行"),
      actionButton("action", "启动选中任务")
    ),
    
    mainPanel(
      DTOutput("output_table")
    )
  )
)

# Run the application
shinyApp(ui = ui, server = server)

关键修改说明 #

• 多进程存储：将单个进程变量改为reactiveValues列表，用行索引作为键，实现多任务独立跟踪，避免新任务覆盖旧任务
• 批量任务支持：修改任务启动逻辑，支持多选行批量启动任务，自动跳过非待执行状态的行
• 全局状态监控：新增每秒触发的观察者，遍历所有后台进程，实时更新任务状态
• 删除逻辑优化：删除行时同步终止对应后台进程，避免无效进程残留
• 交互体验提升：设置DT表格支持多选，方便批量操作

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