基于promises并行运行Shiny模块的实现方法咨询

1. 基础依赖安装与配置 #

• 安装必要工具包：

  install.packages(c("promises", "future"))
  

• 在全局加载脚本（如global/libraries.R）中添加配置：

  library(promises)
  library(future)
  # 设置并行策略：multisession适配本地多核，cluster适用于服务器集群
  plan(multisession, workers = parallel::detectCores() - 1)
  

2. 改造模块服务器函数 #

核心是将模块内**耗时操作（数据库查询、大数据计算等）**包装为Promise对象，让其在后台并行执行。以你的demographics模块为例，改造示例如下：

# 改造前的串行版本（示例）
demographics_server <- function(id, reporter_iso_sel) {
  moduleServer(id, function(input, output, session) {
    # 同步执行的耗时查询
    demo_data <- reactive({
      dbGetQuery(conn, paste0("SELECT * FROM demographics WHERE iso = '", reporter_iso_sel(), "'"))
    })
    
    output$demo_plot <- renderPlot({
      ggplot(demo_data(), aes(x=year, y=population)) + geom_line()
    })
  })
}

# 改造后的异步并行版本
demographics_server <- function(id, reporter_iso_sel) {
  moduleServer(id, function(input, output, session) {
    # 用future_promise将耗时操作丢到并行进程
    demo_data <- reactive({
      future_promise({
        # 每个并行进程需创建独立数据库连接，避免全局连接冲突
        local_conn <- dbConnect(...) # 复用你的DBconnect逻辑
        on.exit(dbDisconnect(local_conn))
        dbGetQuery(local_conn, paste0("SELECT * FROM demographics WHERE iso = '", reporter_iso_sel(), "'"))
      })
    })
    
    # 异步处理结果并渲染输出
    output$demo_plot <- renderPlot({
      demo_data() %>%
        then(function(data) {
          ggplot(data, aes(x=year, y=population)) + geom_line()
        }) %>%
        catch(function(err) {
          showNotification(paste("人口模块加载失败：", err), type = "error")
          ggplot() + annotate("text", x=1, y=1, label="数据加载失败")
        })
    })
  })
}

3. 主服务器中并行启动模块 #

你的现有主服务器是串行调用所有模块，现在可以将无依赖关系的模块用future_promise包装，实现并行启动：

server <- shinyServer(function(input, output, session) {
  # 基础依赖必须串行初始化（所有模块都依赖这些结果）
  mycolor <- color_palette(id = "color")
  reporter_iso_sel <- reporter_download_server(id = "reporterdownload")
  tradeflow <- tradeflow_server(id = "tradeingoods")
  services_tradeflow <- Services_tradeflow_server(id = "tradeinservices")

  # 并行启动独立模块
  future_promise({
    demographics_server(id = "demographics", reporter_iso_sel = reporter_iso_sel$reporter_iso_sel)
  })
  
  future_promise({
    tradeingoods_1_server(
      id = "tradeingoods",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel,
      reporter = reporter_iso_sel$reporter,
      mycolor = mycolor,
      tradeflow = tradeflow
    )
  })
  
  future_promise({
    trade_performance_server(
      id = "tradeperformance",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel,
      reporter = reporter_iso_sel$reporter,
      mycolor = mycolor
    )
  })
  
  future_promise({
    LPI_server(
      id = "LPI",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel,
      reporter = reporter_iso_sel$reporter,
      mycolor = mycolor
    )
  })
  
  future_promise({
    bilateral_server(
      id = "bilateral",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel,
      reporter = reporter_iso_sel$reporter,
      mycolor = mycolor,
      tradeflow = tradeflow
    )
  })
  
  future_promise({
    trade_agreement_server(
      id = "agreement",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel
    )
  })
  
  future_promise({
    services_performance_server(
      id = "servicesperformance",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel,
      reporter = reporter_iso_sel$reporter,
      mycolor = mycolor
    )
  })
  
  future_promise({
    Macroeconomic_server(
      id = "Macroeconomic",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel,
      reporter = reporter_iso_sel$reporter,
      mycolor = mycolor
    )
  })
  
  future_promise({
    Digital_server(
      id = "Digital",
      reporter_iso_sel = reporter_iso_sel$reporter_iso_sel,
      reporter = reporter_iso_sel$reporter,
      mycolor = mycolor
    )
  })
})

4. 核心注意事项 #

• 依赖优先级：如果模块之间存在依赖（如A模块输出是B模块输入），必须保证依赖模块先完成初始化，不能并行这类模块。
• 资源控制：workers参数不要超过CPU核心数，避免系统资源耗尽。
• 连接隔离：每个并行进程需要独立的数据库连接，禁止复用全局连接，建议用连接池（如pool包）管理连接。
• 错误兜底：所有异步操作必须添加catch()捕获错误，避免单个模块崩溃导致整个App挂掉。

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