因此，您有两个要求：功能性要求是您希望异步处理列表中的项，非功能性要求是您不希望同时处理三个以上的项。
关于后者，好的方面是，正如您在问题中已经展示的那样，java本机公开了一个很好的封装 Executor 它以固定大小在线程池上运行任务，允许您在处理线程时优雅地限制并发级别。
转到功能性需求，scala的标准api中包含了一些可以精确实现这一点的功能。特别是它使用 scala.concurrent.Future ，所以为了使用它，我们必须重新构造 triggerAPI 依据 Future . 函数的内容不是特别相关，所以我们现在主要关注它的（修订）签名：

import scala.concurrent.Future
import scala.concurrent.ExecutionContext

def triggerAPI(params: (Int, String))(implicit ec: ExecutionContext): Future[Unit] =
  Future {
    // some code that takes some time to run...
  }

注意现在 triggerAPI 返回一个 Future . 一 Future 可以认为是一个读句柄的东西，将最终计算。特别是，这是一个 Future[Unit] ，在哪里 Unit 代表“我们并不特别关心这个函数的输出，但主要关心它的副作用”。
此外，请注意，该方法现在接受一个隐式参数，即 ExecutionContext . 这个 ExecutionContext 用于提供 Future 在某种形式的环境中进行计算。scala有一个api来创建 ExecutionContext 从 java.util.concurrent.ExecutorService ，因此这对于在固定线程池上运行我们的计算非常方便，在任何给定的时间运行不超过三次回调。
在前进之前，如果你对 Future 是的， ExecutionContext scala文档是您最好的知识来源（这里有几个指针：1，2）。
现在我们有了新的 triggerAPI 方法，我们可以使用 Future.traverse （以下是Scala2.12的文档——撰写本文时的最新版本是2.13，但据我所知，spark用户暂时还停留在2.12上）。
tl；博士 Future.traverse 它采用某种形式的容器和函数，该函数接受容器中的项并返回 Future 其他的东西。该函数将应用于容器中的每个项目，结果将是 Future 结果容器的。在您的情况下：容器是一个 List ，项目为 (Int, String) 而你所归还的其他东西是 Unit .
这意味着你可以这样称呼它：

Future.traverse(inpDS)(triggerAPI)

以及 triggerAPI 将应用于中的每个项目 inpDS .
通过确保线程池支持的执行上下文在调用 Future.traverse ，将使用所需的线程池处理这些项。
通话结果是 Future[List[Unit]] ，这不是很有趣，可以简单地丢弃（因为你只对副作用感兴趣）。
这是一个很大的谈话，如果你想玩我描述的代码，你可以这样做在这里的斯卡斯蒂。
作为参考，这是整个实施过程：

import java.util.concurrent.{ExecutorService, Executors}

import scala.concurrent.duration.DurationLong
import scala.concurrent.Future
import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService}
import scala.util.control.NonFatal
import scala.util.{Failure, Success, Try}

val datasets = List(
  (1, "https://google.com/2X6barD"),
  (2, "https://google.com/3d9vCgW"),
  (3, "https://google.com/2M02Xz0"),
  (4, "https://google.com/2XOu2uL"),
  (5, "https://google.com/2AfBWF0"),
  (6, "https://google.com/36AEKsw"),
  (7, "https://google.com/3enBxz7"),
  (8, "https://google.com/36ABq0x"),
  (9, "https://google.com/2XBjmiF")
)

val executor: ExecutorService = Executors.newFixedThreadPool(3)
implicit val executionContext: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(executor)

def triggerAPI(params: (Int, String))(implicit ec: ExecutionContext): Future[Unit] =
  Future {
    val (index, _) = params
    println(s"+ started processing $index")
    val start = System.nanoTime() / 1000000
    Iterator.from(0).map(_ + 1).drop(100000000).take(1).toList.head // a noticeably slow operation
    val end = System.nanoTime() / 1000000
    val duration = (end - start).millis
    println(s"- finished processing $index after $duration")
  }

Future.traverse(datasets)(triggerAPI).onComplete {
  case result =>
    println("* processing is over, shutting down the executor")
    executionContext.shutdown()
}

你需要关掉电脑 Executor 在你的工作完成之后，它将等待。
尝试添加 pool.shutdown() 你的节目结束了。