在 Scala 和第三方 Java 库中使用 Akka 的最佳实践答案

【问题标题】：Best practices with Akka in Scala and third-party Java libraries在 Scala 和第三方 Java 库中使用 Akka 的最佳实践
【发布时间】：2014-09-25 12:05:05
【问题描述】：

我需要在我的 Scala/Akka 代码中使用 memcached Java API。此 API 为您提供同步和异步方法。异步的返回java.util.concurrent.Future。这里有一个关于在 Scala 中处理 Java 期货的问题 How do I wrap a java.util.concurrent.Future in an Akka Future?。但是在我的情况下，我有两个选择：

使用同步 API 并在未来包装阻塞代码并标记阻塞：

Future {
  blocking {
    cache.get(key) //synchronous blocking call
  } 
}

使用异步 Java API 并在 Java Future 上每隔 n 毫秒轮询一次，以检查未来是否完成（如上述链接问题中的上述答案之一所述）。

哪个更好？我倾向于第一个选项，因为轮询会极大地影响响应时间。 blocking { } block 不应该阻止阻塞整个池吗？

【问题讨论】：

标签： java scala asynchronous concurrency akka

【解决方案1】：

我总是选择第一个选项。但我做的方式略有不同。我不使用blocking 功能。（实际上我还没有考虑过。）相反，我为 Future 提供了一个自定义执行上下文，它包装了同步阻塞调用。所以它看起来基本上是这样的：

val ecForBlockingMemcachedStuff = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(100)) // whatever number you think is appropriate
// i create a separate ec for each blocking client/resource/api i use

Future {
    cache.get(key) //synchronous blocking call
}(ecForBlockingMemcachedStuff) // or mark the execution context implicit. I like to mention it explicitly.

因此所有阻塞调用都将使用专用的执行上下文（= Threadpool）。因此，它与负责非阻塞内容的主要执行上下文分开。

Typesafe 提供的online training video for Play/Akka 中也解释了这种方法。第 4 课中有一个关于如何处理阻塞呼叫的视频。 Nilanjan Raychaudhuri 解释了这一点（希望我拼写正确），他是 Scala 书籍的知名作者。

更新：我有一个discussion with Nilanjan on twitter。他解释了使用blocking 和自定义ExecutionContext 的方法之间的区别是什么。 blocking 功能只是创建了一个特殊的ExecutionContext。它提供了一种简单的方法来解决您需要多少线程的问题。当池中所有其他现有线程都忙时，它每次都会产生一个新线程。所以它实际上是一个不受控制的 ExecutionContext。它可能会创建大量线程并导致出现内存不足错误等问题。因此，具有自定义执行上下文的解决方案实际上更好，因为它使这个问题变得明显。 Nilanjan 还补充说，您需要考虑在此池因请求超载的情况下进行断路。

TLDR：是的，阻塞呼叫很糟糕。 使用自定义/专用 ExecutionContext 来阻止调用。还要考虑断路。

【讨论】：

是的，我认为这是第三种方法。我只是想知道我是否可以使用自定义执行上下文和blocking 功能的组合。我在某处读到，如果您使用自定义执行上下文，此功能无效，因此在悲观的情况下，您可以阻止自定义池。

【解决方案2】：

Akka documentation 提供了一些关于如何处理阻塞调用的建议：

在某些情况下，做阻塞操作是不可避免的，即把一个线程休眠一个不确定的时间，等待外部发生的事件。示例是旧的 RDBMS 驱动程序或消息传递 API，根本原因通常是（网络）I/O 发生在封面。面对这种情况时，您可能很想将在 Future 中阻塞调用并改为使用它，但是这个策略太简单了：你很可能会发现瓶颈或当应用程序运行时内存或线程不足增加加载。

“阻塞”的适当解决方案的非详尽列表问题”包括以下建议：

在一个参与者（或由路由器管理的一组参与者）内执行阻塞调用，确保配置一个线程池，该线程池是专门用于此目的或足够大。

在 Future 内执行阻塞调用，确保在任何时间点此类调用的数量都有上限（提交无限制的这种性质的任务数量会耗尽你的内存或线程限制）。

在 Future 中执行阻塞调用，为线程池提供一个线程数上限，该上限适合于运行应用程序的硬件。

专用单个线程来管理一组阻塞资源（例如，驱动多个通道的 NIO 选择器）并在事件发生时分派事件作为参与者消息出现。

第一种可能性特别适合本质上是单线程的，例如数据库句柄传统上一次只能执行一个未完成的查询并使用内部同步以确保这一点。一个常见的模式是创建 N 个参与者的路由器，每个参与者包装一个 DB 连接和处理发送到路由器的查询。然后必须调整数字 N 最大吞吐量，这取决于哪个 DBMS 部署在什么硬件上。

【讨论】：