基于可用 FREE cpu 的 Java 并发性

Question

问题

当且仅当有空闲 CPU 时，我如何扩展以使用更多线程？像 ThreadPoolExecutor 这样的东西，它在 cpu 核心空闲时使用更多线程，如果没有空闲则更少或只使用一个线程。

用例

现状:我的 Java 服务器应用程序处理请求并提供结果。有一个 ThreadPoolExecutor 以合理数量的最大线程为请求提供服务，遵循以下原则:cpu 核心数 = 最大线程数。执行的工作是 cpu 繁重的，并且有一些磁盘 IO (DB)。代码是线性的，单线程的。处理单个请求需要 50 到 500 毫秒。有时每分钟只有几个请求，有时同时有 30 个请求。具有 12 个内核的现代服务器可以很好地处理负载。吞吐量不错，延迟还可以。

期望的改进:当请求数量较少时(大多数情况下都是这种情况)，许多 cpu 内核处于空闲状态。在这种情况下，可以通过为单个请求多线程运行一些代码来改善延迟。一些原型(prototype)设计显示出改进，但一旦我用更多的并发请求进行测试，服务器发疯了。吞吐量下降，内存消耗过多。30 个并发请求共享一个 10 个队列，这意味着 20 个等待时最多可以运行 10 个，并且这 10 个中的每一个同时使用多达 8 个线程来实现并行性，对于一台机器来说似乎太多了有 12 个内核(其中 6 个是虚拟内核)。

这在我看来是一个常见的用例，但我无法通过搜索找到信息。

想法

1) 请求计数一种想法是计算当前已处理请求的数量。如果为 1 或低，则执行更多并行操作，如果高则不做任何事情并像以前一样继续单线程。这听起来很容易实现。缺点是:请求计数器重置不能包含错误，最后想想而且它实际上并不检查可用的 cpu，也许另一个进程也使用 cpu。在我的例子中，机器专用于此应用程序，但仍然如此。

2) 实际cpu查询我认为正确的方法是只询问 cpu，然后再决定。由于 Java7 有 OperatingSystemMXBean.getSystemCpuLoad() 参见 http://docs.oracle.com/javase/7/docs/jre/api/management/extension/com/sun/management/OperatingSystemMXBean.html#getSystemCpuLoad()但我找不到任何提到 getSystemCpuLoad 和 ThreadPoolExecutor 或类似内容的网页关键字的组合，这告诉我这不是一条好路要走。JavaDoc 说“返回整个系统的‘最近 cpu 使用率’”，我想知道是什么“最近的 CPU 使用率”是指最近的时间以及该调用的费用。

更新

我将这个问题搁置了一段时间，看看是否有更多的意见。没有。虽然我不喜欢对技术问题的“不行”的回答，但我现在要接受 Holger 的回答。他名声好，论据好，其他人都认可他的回答。我自己对想法 2 进行了一些试验。我查询了任务中的 getSystemCpuLoad() 来决定他们自己的 ExecutorService 有多大。正如 Holger 所写，当有一个 SINGLE ExecutorService 时，可以很好地管理资源。但是一旦任务开始自己的任务，它们就不能 - 这对我来说没有成功。

Answer 1

没有基于“空闲 CPU”的限制方式，无论如何它都行不通。有关“免费 CPU”的信息在您获得后即已过时。假设你有 12 个并发运行的线程，同时检测到有一个空闲的 CPU 内核并决定安排一个子任务……

您可以做的是限制最大资源消耗，这在为 所有 任务使用最大线程数的单个 ExecutorService 时效果很好。

棘手的部分是任务对子任务结果的依赖性，这些子任务在稍后排队并且由于工作线程数量有限可能仍处于待处理状态。

如果任务检测到其子任务仍未决，则可以通过撤销并行执行来调整这一点。为此，请手动为子任务创建一个 FutureTask 并使用 execute 而不是 submit 安排它。然后在任务中正常进行，并在顺序执行中执行子任务的地方检查是否可以 remove ThreadPoolExecutor 中的 FutureTask。与 cancel 不同，这仅在尚未启动时有效，因此表明没有空闲线程。因此，如果 remove 返回 true，您可以就地执行子任务，让所有其他线程执行任务而不是子任务。否则，您可以等待结果。

在这个地方值得注意的是，如果任务适应 I/O 操作(或者可能等待子任务)，线程数多于 CPU 内核是可以的。这里的重点是一个限制。

FutureTask<Integer> coWorker = new FutureTask<>(/* callable wrapping sub-task*/);
executor.execute(coWorker);

// proceed in the task’s sequence

if(executor.remove(coWorker)) coWorker.run();// do in-place if needed
subTaskResult=coWorker.get();

// proceed