Windows系统下如何用Plumber实现多请求并行处理？

问题背景 #

你尝试通过Plumber结合future库实现多请求并行处理，该方案在Linux系统中可正常工作，但在Windows系统下所有请求串行执行，返回的PID完全相同，无法实现并行计算。你的测试代码如下：

Plumber服务代码 #

# ./plumber.R
library(future)
library(promises)
# fix the number of workers to 5
future::plan(future::multisession(workers = 5))

#' @param msg The message to echo back.
#' @serializer text
#' @get /echo
function(msg = ""){
  future({
    Sys.sleep(5)
    paste0(Sys.getpid(), " - ", as.character(Sys.time()), "\n")
  })
}

启动服务代码 #

library(plumber)
pr("plumber.R") %>% pr_run(port = 8080)

测试代码 #

library(httr)
responses <- lapply(1:6, function(i) {
  GET("http://127.0.0.1:8080/echo")
})

print(responses)

解决方案 #

Windows系统下完全可以实现Plumber的并行处理，问题出在Plumber的异步模式未启用，以及测试方式的串行特性。以下是两种可行的实现方案：

方案1：修复future+promises的实现 #

Plumber v1.0+支持异步请求处理，但需要显式启用。同时，测试代码需改为异步发送请求，避免串行调用导致的假象。

修正后的Plumber服务代码

# ./plumber.R
library(future)
library(promises)

# Windows下使用multisession计划，设置5个工作进程
future::plan(future::multisession(workers = 5))

#' @param msg The message to echo back.
#' @serializer text
#' @get /echo
function(msg = ""){
  # 使用promises管道操作符确保返回可被Plumber识别的异步对象
  future({
    Sys.sleep(5)
    paste0(Sys.getpid(), " - ", as.character(Sys.time()), "\n")
  }) %...>% identity()
}

启用异步模式启动服务

library(plumber)
# 关键：设置async=TRUE启用异步处理
pr("plumber.R") %>% pr_run(port = 8080, async = TRUE)

异步测试代码

避免用lapply串行发送请求，改用异步请求方式：

library(httr)
library(promises)

# 异步发送6个请求
async_requests <- lapply(1:6, function(i) {
  async_GET("http://127.0.0.1:8080/echo")
})

# 等待所有请求完成并获取结果
resolved_responses <- resolve_all(async_requests)
lapply(resolved_responses, function(resp) content(resp, "text"))

执行后你会看到不同的PID和分散的时间戳，证明并行处理生效。

方案2：使用doParallel+foreach实现 #

也可以通过doParallel配合foreach实现并行，同样需要结合Plumber的异步模式：

Plumber服务代码

# ./plumber.R
library(doParallel)
library(foreach)
library(promises)
library(future)

# 注册并行集群（Windows下用makeCluster）
cl <- makeCluster(5)
registerDoParallel(cl)

#' @param msg The message to echo back.
#' @serializer text
#' @get /echo
function(msg = ""){
  # 将foreach并行任务包装为future对象，返回异步promise
  future({
    foreach(i = 1, .combine = "c") %dopar% {
      Sys.sleep(5)
      paste0(Sys.getpid(), " - ", as.character(Sys.time()), "\n")
    }
  }) %...>% identity()
}

启动服务与测试

同样需要用pr_run(async = TRUE)启动服务，测试代码与方案1的异步测试代码一致。

关键注意事项 #

1. Windows进程模型限制：Windows不支持fork，因此必须使用multisession（future）或makeCluster（doParallel）创建独立工作进程，不能使用multicore计划。
2. 异步模式必须启用：Plumber默认同步运行，需通过pr_run(async = TRUE)开启异步支持，才能处理返回promise的路由函数。
3. 测试方式要匹配：串行的测试调用（如lapply+GET）无法体现并行效果，必须用异步请求或多进程同时发送请求。

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