问题标签 [completion-service]

假设您正在编写如下代码：

如果您看到上面的代码，有没有办法找出在我调用completion.poll 时返回的可调用对象并有未来作为回报？

我可以扩展 Callable 以获得某种 ID 来做到这一点，但我想知道 java 本身是否有可能。

我使用 ExecutorCompletionService 来调用 Web 服务。线程正在访问的 Web 服务会在 5 秒内返回响应。但是，completionService.take().get() 需要将近 30 秒才能从其第一个完成的线程中检索结果；而 completionService.take().get() 需要 6 秒才能从其他已完成线程中检索结果。

我知道 ExecutorCompletionService 它需要（）.get（）一个首先完成的任务，而不是首先提交的任务。但我无法弄清楚为什么要花这么多时间（30 秒+）才能从第一个完成的线程中获得结果。为什么其他线程在控制台中声称他们需要 6 秒才能完成，但 take().get() 不会比需要 30+ 秒的线程先处理。

最初，我开始仅将 ExecutorService 与 Future 一起使用，但后来我切换到 ExecutorCompletionService 以获得更好的响应时间。但响应时间不会改变。除了我的主要代码之外，还尝试使用 ExecutorCompletionService 在 Future 中添加任务，但没有任何改变：

系统输出控制台的响应时间（以秒为单位）：

请帮助我纠正我做错的任何事情。我只知道，如果 4 个线程使用 Executor Completion Service 并行运行，而另一个服务被击中一个线程的时间不超过 5-6 秒，那么我的第一个完成的线程不应该超过 6-7 秒（只是像其他线程一样）。

这导致对用户的总响应时间几乎是 33 秒。

假设我有一个长时间运行的操作，如果需要太长时间，我想从中获得结果或超时。

我想比较两个不同的版本：

1）当我们调用时，poll我们检查是否有任何项目被排入内部阻塞队列

2）当我们调用时，get我们有FutureTask等待的实现

基本上在这两个版本中，我们都在等待未来完成它的执行，但是内部实现是不同的。在性能方面会有什么不同吗？我已经看到了这个答案，它与 https://stackoverflow.com/a/52980559/2055854 进行比较CompletionService并说CompletableFuture应该有。执行上会一样吗？如果是的话，很高兴听到更好的解释。FutureTask

我正在开发具有以下一般架构的 Java 服务器应用程序：

• 客户端向服务器发出 RPC 请求
• 我相信 RPC 服务器（gRPC）有自己的线程池来处理请求
• 请求立即插入以Thread Pool 1进行更多处理
• 一个特定的请求类型，我们称之为Request R，需要并行运行几个异步任务，判断结果形成共识，返回给客户端。这些任务运行时间更长一些，所以我使用单独Thread Pool 2的来处理这些请求。重要的是，每个Request R都需要运行相同的 2-3 个异步任务。Thread Pool 2因此服务所有当前正在执行Request R的。但是，aRequest R应该只能查看和检索属于它的异步任务。
• 为了实现这一点，在每次传入Request R时，当它进入时Thread Pool 1，它将CompletionService为请求创建一个新的，由Thread Pool 2. 它将提交 2-3 个异步任务，并检索结果。这些应该与可能运行在Thread Pool 2属于其他请求的任何其他内容严格隔离。
• 我的问题：
  • 首先，Java 是CompletionService孤立的吗？检查JavaDocs后，我找不到关于此的好的文档。换句话说，如果两个或多个CompletionService's 由同一个线程池支持，它们中的任何一个是否有可能将未来拉到另一个s 上CompletionService？
  • CompletionService其次，为每个请求创建这么多的这是不好的做法吗？有没有更好的方法来处理这个？当然，为每个请求创建一个新的线程池是一个坏主意，那么有没有更规范/正确的方法来隔离 a 中的期货，CompletionService或者我正在做什么？

在此先感谢您的帮助。任何指向有用的文档或示例的指针将不胜感激。

代码，供参考，虽然微不足道：

我有 1000 个大文件要按如下所述顺序处理：

1. 首先需要将这些文件并行复制到不同的目录，我计划使用ExecutorService10 个线程来实现它。
2. 只要将任何文件复制到另一个位置（#1），我就会将该文件提交给ExecutorService10 个线程进行进一步处理。
3. 最后，需要对这些文件并行执行另一个操作，例如 #2 从 #1 获取输入，#3 从 #2 获取输入。

现在，我可以CompletionService在这里使用，所以我可以按照它们完成的顺序处理从 #1 到 #2 和 #2 到 #3 的线程结果。CompletableFuture说我们可以将异步任务链接在一起，这听起来像是我可以在这种情况下使用的东西。

我不确定我是否应该使用CompletableFuture（因为它相对较新并且应该更好）来实施我的解决方案，或者是否CompletionService足够？在这种情况下，我为什么要选择一个而不是另一个？

问题

我正在使用完成服务并产生子线程来执行一些 ETL。当我在 IDE 中调试然后停止所有进程时，我注意到我仍然有一堆僵尸线程正在杀死我的 CPU。这是因为我没有正确终止子线程。

最小示例

想法

本质上，我将我的提交Callable<Boolean>给我的完成服务。

如果我停止调试器，该线程可能仍在运行。我刚刚添加了future.cancel(true)这块，它似乎已经停止从我的子线程连续上传文件，但我仍然看到这些 java 进程在我的活动监视器上运行。

我想知道我应该如何考虑这个问题？我想要可调用的，因为它告诉我底层 ETL何时完成（真/假）

编辑：future.cancel 实际上似乎有很大帮助..这是我想要使用的吗？

我正在并行调用一些方法。如果任何方法抛出异常，我想使用传递给方法的值。

PS：请忽略任何语法错误。

我需要知道当您提交的任务多于当前池大小时，Java 是如何处理的。我有以下代码：

如果提交是阻塞调用，有什么线索吗？或者如果线程都忙，它会拒绝任务吗？

谢谢你们。

我正在编写一个程序，它使用 aCompletionService在一堆不同的对象上运行线程分析，其中每个“分析”都包括接收一个字符串并进行一些计算以给出true或false作为答案。我的代码看起来基本上是这样的：

在标记部分，我的目标是在超过 300 个字符串失败时让它放弃计算，这样我就可以继续进行新的分析，用一些不同的数据再次调用这个方法。问题是，由于CompletionService再次使用相同的，如果我不以某种方式清除队列，那么工作队列将继续增长，因为我每次使用它时都会不断添加更多（因为在 300 次失败之后，可能仍有许多未处理左字符串）。

我尝试使用类似的方法遍历futures列表并删除所有未完成的任务futures.foreach(future -> future.cancel(true)，但是当我下次调用该方法java.util.concurrent.CancellationException时，当我尝试调用时出现错误resFuture.get()。

（编辑：似乎即使我调用foreach(future->future.cancel(true))，但这并不能保证workerQueue之后实际上是清除的。我不明白为什么会这样。似乎清除队列需要一段时间，并且代码不会等待以便在进行下一次分析之前发生这种情况，因此偶尔get会在已取消的未来上调用。）

我也尝试过

清空队列，虽然这行得通，但它几乎不比只运行所有分析快，因此它对效率的影响并不大。

我的问题是是否有更好的方法来清空工作队列，这样当我下次调用此代码时，它就好像CompletionService又是新的了。

编辑：我尝试过的另一种方法只是设置executorCompletionService = new CompletionService，它比我的其他解决方案略快，但仍然相当慢，绝对不是好习惯。

PS：也很高兴接受任何其他可能的方式，我不喜欢使用CompletionService它只是迄今为止我所做的最简单的事情