Scala Play框架中基于URL列表动态发起并行REST调用

我来帮你搞定这个动态批量调用的需求！原来的示例是固定几个URL的串行调用（for推导式里的请求是一个接一个执行的），要实现从URL列表动态处理，咱们可以借助Future.sequence实现并行批量请求，或者用foldLeft实现串行请求，下面分情况给你说明：

核心思路：从URL列表到响应列表 #

首先，你需要把URL列表转换成Future[WSResponse]的集合，然后通过Future.sequence把这个集合转换成Future[List[WSResponse]]——这个方法的作用是把“一堆Future”变成“包含一堆结果的Future”，这样就能一次性等待所有请求完成，拿到所有响应的列表。

并行批量请求（推荐，效率更高） #

如果你的场景允许同时发起所有请求，并行执行是最优选择。下面是完整的示例代码（假设你在控制器中注入了WSClient）：

import play.api.libs.ws._
import play.api.mvc._
import scala.concurrent.ExecutionContext

class BatchCallController @Inject()(ws: WSClient, ec: ExecutionContext) extends BaseController {

  // 示例：处理传入的URL列表，返回处理后的结果
  def processUrls(urls: List[String]) = Action.async {
    // 1. 把每个URL转换成Future[WSResponse]
    val futureResponses: List[Future[WSResponse]] = urls.map { url =>
      ws.url(url).get()(ec)
    }

    // 2. 把Future的列表转换成包含响应列表的Future
    val allResponses: Future[List[WSResponse]] = Future.sequence(futureResponses)(ec)

    // 3. 处理所有响应
    allResponses.map { responses =>
      // 这里可以自定义处理逻辑，比如遍历每个响应解析内容
      val processedResults = responses.map { response =>
        s"URL: ${response.uri} | Status: ${response.status} | Body snippet: ${response.body.take(50)}"
      }
      Ok(processedResults.mkString("\n"))
    }
  }
}

串行批量请求（和原示例行为一致） #

如果你需要和原示例一样，一个请求完成后再发起下一个（比如有依赖关系或者需要控制请求频率），可以用foldLeft来串行累积结果：

def processUrlsSerial(urls: List[String]) = Action.async {
  // 初始值是一个包含空列表的Future
  val initialFuture: Future[List[WSResponse]] = Future.successful(Nil)(ec)

  // 逐个发起请求，把响应添加到列表中
  val serialResponses: Future[List[WSResponse]] = urls.foldLeft(initialFuture) { (accFuture, url) =>
    accFuture.flatMap { acc =>
      ws.url(url).get()(ec).map(response => acc :+ response)
    }
  }

  serialResponses.map { responses =>
    // 同样处理响应列表
    val processed = responses.map(r => s"${r.uri} -> ${r.status}")
    Ok(Json.toJson(processed))
  }
}

处理部分请求失败的情况 #

如果担心某个请求失败导致整个批量任务失败，可以给每个Future添加错误处理，用Either来包裹结果：

def processUrlsWithErrorHandling(urls: List[String]) = Action.async {
  val futureResponses: List[Future[Either[Throwable, WSResponse]]] = urls.map { url =>
    ws.url(url).get()(ec)
      .map(Right(_)) // 请求成功时包装成Right
      .recover { case e => Left(e) } // 请求失败时包装成Left
  }

  val allResponses: Future[List[Either[Throwable, WSResponse]]] = Future.sequence(futureResponses)(ec)

  allResponses.map { results =>
    val processed = results.map {
      case Right(response) => s"Success: ${response.uri} (${response.status})"
      case Left(error) => s"Failed: ${error.getMessage}"
    }
    Ok(processed.mkString("\n"))
  }
}

关键知识点说明 #

• Future.sequence：并行执行所有Future，等待全部完成后返回结果列表，只要有一个Future失败，整个结果Future就会失败（除非你提前处理了每个Future的错误）。
• 依赖注入：Play 2.6+版本中，WSClient需要通过依赖注入获取，不要直接使用WS.url()（旧版本的静态方法已经废弃）。
• ExecutionContext：尽量注入专用的ExecutionContext，避免使用全局的ExecutionContext.Implicits.global，这样能更好地控制线程池资源。

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