函数内更新Shiny响应式变量失败，求解决方案

背景 #

我用future包实现了并行化for循环，需要把结果写入CSV文件。为了避免写入冲突这类问题，我把循环拆成了两部分：内层并行循环负责生成结果，外层循环负责写入结果。

需求 #

我想展示外层循环的进度，于是创建了响应式变量react_val，并尝试用ipc包的shinyQueue函数在自定义函数里更新这个变量，但更新操作完全没生效，想知道怎么解决。

原代码 #

library(shiny); library(ipc); library(future); plan(multisession)

ftc_parallel <- function(queue){
    start<- Sys.time()
    for(i in 1:5){
        command_set <- vector(mode = "list")
        for(j in 1:3){
            command_set[[j]] <- future({
                Sys.sleep(sample(1:3, size = 1))
                df <- data.frame(
                    Colonne1 = runif(2),
                    Colonne2 = runif(2),
                    Colonne3 = runif(2)
                )
                return(df)
            }, seed = T)

        }
        queue$producer$fireAssignReactive("react_val", i)
        # write in file
        # results <- bind_rows(Filter(Negate(is.null), lapply(command_set, value)))
        # write.table(results, append = T, sep=";")
    }
    end <- Sys.time()
    time <- end - start
    print(time)

}


ui <- fluidPage(titlePanel(""), sidebarLayout( sidebarPanel(), mainPanel(
    actionButton("bouton", "Button")
)))

server <- function(input, output) {

    react_val <- reactiveVal(0)
    queue <- shinyQueue()
    queue$consumer$start(100)

    observe({ print(react_val())})

    observeEvent(input$bouton, {
        ftc_parallel(queue)
    })
}

shinyApp(ui = ui, server = server)

问题原因 #

直接在observeEvent里同步调用ftc_parallel会阻塞Shiny的主进程，导致shinyQueue的消费者无法及时处理更新信号，自然就看不到react_val的变化。另外，创建shinyQueue时没绑定session上下文，也会让信号传递出问题。

解决办法 #

把ftc_parallel的执行放到异步任务里，同时确保队列和session绑定，具体修改后的代码如下：

library(shiny); library(ipc); library(future); plan(multisession)

ftc_parallel <- function(queue, session){
    start<- Sys.time()
    for(i in 1:5){
        command_set <- vector(mode = "list")
        for(j in 1:3){
            command_set[[j]] <- future({
                Sys.sleep(sample(1:3, size = 1))
                df <- data.frame(
                    Colonne1 = runif(2),
                    Colonne2 = runif(2),
                    Colonne3 = runif(2)
                )
                return(df)
            }, seed = T)
        }
        # 传入session上下文，确保信号能正确传递
        queue$producer$fireAssignReactive("react_val", i, session = session)
        # 给主进程留一点处理事件的时间，让进度更新更流畅
        Sys.sleep(0.1)
        # 恢复写入文件的代码（这里补充了文件存在时不重复写表头的逻辑）
        # results <- bind_rows(Filter(Negate(is.null), lapply(command_set, value)))
        # write.table(results, append = TRUE, sep=";", file = "output.csv", row.names = FALSE, col.names = !file.exists("output.csv"))
    }
    end <- Sys.time()
    time <- end - start
    print(time)
}

ui <- fluidPage(
    titlePanel("并行任务进度展示"), 
    sidebarLayout( 
        sidebarPanel(), 
        mainPanel(
            actionButton("bouton", "启动并行任务"),
            # 新增进度显示文本，直观展示进度
            textOutput("progress_text")
        )
    )
)

server <- function(input, output, session) {
    react_val <- reactiveVal(0)
    # 创建队列时绑定当前session
    queue <- shinyQueue(session = session)
    queue$consumer$start(100)

    # 渲染进度文本
    output$progress_text <- renderText({
        paste0("当前进度：", react_val(), "/5")
    })

    observeEvent(input$bouton, {
        # 把并行函数的执行放到future里，避免阻塞主进程
        future({
            ftc_parallel(queue, session)
        })
    })
}

shinyApp(ui = ui, server = server)

关键修改点 #

1. 异步执行任务：把ftc_parallel的调用包裹在future中，让Shiny主进程能同时处理UI事件和队列信号。
2. 绑定session上下文：创建shinyQueue和调用fireAssignReactive时都传入session参数，确保信号能准确传递到对应的响应式变量。
3. 添加进度UI：新增textOutput来直观显示进度，代替原来的控制台打印。
4. 预留事件处理时间：添加Sys.sleep(0.1)给主进程留出处理事件的间隙，让进度更新更流畅。

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