Shiny异步编程使用{crew}遇维度不匹配错误求助

需求场景 #

Shiny应用教师端支持通过selectizeInput选择学生ID，从Firestore获取数据并绘制ggiraph交互式图表，图表点可点击弹出模态框展示详情。

面临挑战 #

10-15位教师同时使用，单学生图表绘制耗时≥2秒，常规Shiny并发体验差，期望通过{crew}在独立R会话中异步绘图，实现主应用会话+独立绘图会话的架构。

错误问题 #

并发测试5-8用户时出现以下错误，导致部分用户断开连接：

Warning: Error in attributes<-: dims [product 5] do not match the length of object [1]
63: query_status
62: mirai::status
61: <Anonymous>
57: daemons_info
54: private$.launcher$scale
53: .subset2(self, "scale")
52: controller()$pop
51: observe
50: <observer>
3: runApp
2: print.shiny.appobj
1: <Anonymous>

Warning: Error in attributes<-: dims [product 5] do not match the length of object [1]
59: query_status
58: mirai::status
57: <Anonymous>
53: daemons_info
52: controller()$client$summary
51: observe
50: <observer>
3: runApp
2: print.shiny.appobj
1: <Anonymous>

Warning: Error in attributes<-: dims [product 5] do not match the length of object [1]
63: query_status
62: mirai::status
61: <Anonymous>
57: daemons_info
55: self$tally
54: private$.launcher$scale
53: .subset2(self, "scale")
52: controller()$pop
51: observe
50: <observer>
3: runApp
2: print.shiny.appobj
1: <Anonymous>

代码问题诊断 #

你的{crew}实现存在以下核心问题：

1. 每个用户会话创建独立crew控制器 #

在模块内为每个用户实例化crew_controller_local(workers=1)，10-15个用户会创建10-15个独立控制器，每个控制器至少启动1个worker进程。过多的控制器和进程会导致系统资源竞争，mirai底层进程通信时出现维度不匹配的错误（即你看到的dims [product 5] do not match）。

2. 全局变量导致控制器实例混乱 #

使用curr_controller <<-全局赋值控制器，在Shiny多用户环境中，不同用户的模块会覆盖这个全局变量，导致：

• 一个用户终止会话时错误关闭其他用户的控制器
• 查询任务状态时访问错误的控制器实例，引发状态查询异常

3. 重复任务推送与队列积压 #

reactivePoll每10秒触发一次，每次都会检查缓存并推送新任务，即使队列中已有相同学生ID+数据哈希的任务。任务积压会导致控制器状态查询时出现数据维度不匹配的错误。

4. 任务command参数传递错误 #

推送任务时，command参数直接传入了Shiny reactive对象（event_names_list_stu、accessToken、uid），虽然data参数里传了对应的值，但command中应该引用data里的变量，而非reactive对象本身，可能导致任务执行时上下文错误。

优化方案建议 #

1. 全局共享crew控制器 #

在应用启动时创建一个全局的crew控制器，所有用户共享，避免重复创建控制器和进程：

# 在app.R全局环境中初始化控制器
global_controller <- crew_controller_local(
  workers = 8,  # 根据并发数设置，建议等于预期最大用户数的60%-80%
  seconds_idle = 30  # 空闲worker自动终止的时间，避免资源浪费
)
global_controller$start()

# 应用停止时终止控制器
onStop(function() {
  global_controller$terminate()
})

2. 修正模块内控制器使用逻辑 #

模块内直接使用全局控制器，去掉模块内的控制器创建逻辑：

# 模块server中移除原有的controller() reactive和curr_controller变量
# 直接使用global_controller

observeEvent(event_names_list_stu(), {
  req(uid())
  student_id <- uid()
  data_hash <- current_data_hash()

  cached_result <- cache_manager$get_cached_plot(student_id, data_hash)
  if (!is.null(cached_result)) {
    reactive_result(cached_result)
    reactive_status(paste(format(Sys.time()), "Retrieved from cache 📋"))
    return()
  }

  # 检查队列中是否已有相同任务（避免重复推送）
  existing_tasks <- global_controller$client$summary()
  task_exists <- any(existing_tasks$description == paste(student_id, data_hash, sep = "-"))

  if (!task_exists) {
    reactive_result(NULL)
    global_controller$push(
      command = create_inst_progress_visual(event_names_list_stu, accessToken, uid),
      data = list(
        create_inst_progress_visual = create_inst_progress_visual,
        event_names_list_stu = event_names_list_stu(),
        accessToken = accessToken(),
        uid = uid()
      ),
      packages = c("ggplot2", "ggiraph", "dplyr", "httr2", "frstore", "ggimage", "here"),
      description = paste(student_id, data_hash, sep = "-")  # 添加任务标识，用于去重
    )
    reactive_poll(TRUE)
  }
})

3. 优化reactivePoll逻辑 #

让checkFunc返回数据的哈希值，只有当数据真正变化时才触发valueFunc，避免无意义的任务检查：

event_names_list_stu <- reactivePoll(
  10000,
  session,
  checkFunc = function() {
    # 返回数据哈希，只有哈希变化时才更新valueFunc
    data <- get_all_event_names_for_a_student(collection_paths_e, uid(), accessToken())
    digest::digest(data)
  },
  valueFunc = function() {
    get_all_event_names_for_a_student(collection_paths_e, uid(), accessToken())
  }
)

4. 任务错误捕获 #

在绘图函数中加入错误捕获，避免单个任务失败导致控制器状态异常：

create_inst_progress_visual <- function(event_names_list_stu, accessToken, uid) {
  tryCatch({
    # 原绘图逻辑
    list(
      progress_plot = plot,
      status = status,
      dat = dat
    )
  }, error = function(e) {
    message("绘图任务失败: ", e$message)
    list(
      progress_plot = NULL,
      status = NULL,
      dat = NULL,
      error = e$message
    )
  })
}

5. 优化任务状态查询逻辑 #

避免频繁调用controller()$client$summary()，减少状态查询的开销：

observe({
  req(reactive_poll())
  invalidateLater(millis = 500)  # 延长轮询间隔，减少资源消耗
  task <- global_controller$pop()

  if (!is.null(task)) {
    if ("result" %in% names(task) && !is.null(task$result)) {
      if ("progress_plot" %in% names(task$result[[1]]) && !is.null(task$result[[1]]$progress_plot)) {
        progress_map <- task$result[[1]]$progress_plot
        statuz <- task$result[[1]]$status
        datz <- task$result[[1]]$dat

        cache_manager$update_cache(uid(), progress_map, current_data_hash())
        reactive_result(progress_map)
        status(statuz)
        dat(datz)
      } else if ("error" %in% names(task$result[[1]])) {
        reactive_status(paste(format(Sys.time()), "绘图失败: ", task$result[[1]]$error))
      }
    }
  }

  # 只在有任务时更新状态
  task_count <- sum(global_controller$client$summary()$assigned) - sum(global_controller$client$summary()$complete)
  if (task_count > 0) {
    reactive_status(status_message(n = task_count))
    reactive_poll(TRUE)
  } else {
    reactive_status(paste(format(Sys.time()), "无待处理任务"))
    reactive_poll(FALSE)
  }
})

额外建议 #

• 缓存机制优化：使用分布式缓存（如Redis）替代本地内存缓存，支持多进程/多实例共享缓存，避免不同worker重复绘制相同图表。
• 监控与日志：添加任务执行日志，便于排查并发场景下的错误。
• 资源限制：根据服务器CPU核心数设置worker数量，避免过度占用资源（一般worker数不超过CPU核心数的80%）。

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