问题标签 [completion-service]

我刚刚在这篇博文中找到了 CompletionService 。但是，这并没有真正展示 CompletionService 相对于标准 ExecutorService 的优势。可以用其中任何一个编写相同的代码。那么，CompletionService 什么时候有用呢？

您能否提供一个简短的代码示例以使其一目了然？例如，此代码示例仅显示不需要 CompletionService 的位置（=等效于 ExecutorService）

所以，我对多线程还很陌生，并且最近在我的所有程序中都使用了这个想法。在我开始更多地使用它之前，我真的想确保它是使用 Executor、CompletionService 和 BlockingQueue 以及 Observer 实现多线程的正确有效方法。我将在下面提供示例代码，但首先让我快速解释一下我认为它是如何工作的，也许这会有所帮助。

我拥有的第一件事是 BlockingQueue，所有任务都通过 add(Task task) 方法添加到此队列中。在创建类时，调用 run 方法并调用 while(true) 调用队列阻塞，直到将某些内容添加到任务队列中。

一旦某些东西被添加到 run() queue.take() 内的队列中，就会返回队列中的项目。然后我拿那个项目并将它传递给 WorkerThread 类，该类在它上面做一些事情。该 workerThread 被添加到处理等待线程完成的 CompletionService 池中。

好的，现在是我不确定是否正确的部分。我还有一个实现可运行的内部类，并在类初始化时启动。它的工作是永远循环调用 pool.take()。因此，这实质上是等待其中一个 WorkerThread 完成。我让完成服务处理这个。一旦 take() 得到一个值，内部类将它传递给一个通知观察者方法。

这可以实现吗？让我有点担心的是，主类在任务队列上运行了一个 while(true) 循环，而一个内部类也在循环等待池以接收来自 WorkerThread 的结果？

这是一个示例实现。你认为呢？

我正在开发应用程序，应该从 db 获取用户列表并从目录（ldap 或 AD）更新您的详细信息。我在多核机器上执行此过程，所以我创建了这个应用程序（下面的代码）。我正在使用 CompletionService 并在 Future 对象中获取结果。

一段时间后，我出现“无法创建新的本机线程”消息的内存不足错误。在任务管理器中，我看到该应用程序创建了大量线程，但我要求创建大小等于处理器数量的固定线程池。

我的代码有什么问题？

错误 CheckGroupMembership：85 - java.lang.OutOfMemoryError：无法创建新的本机线程 java.util.concurrent.ExecutionException：java.lang.OutOfMemoryError：无法在 java.util.concurrent.FutureTask$Sync.innerGet 处创建新的本机线程（ FutureTask.java:222) 在 java.util.concurrent.FutureTask.get(FutureTask.java:83

GetuserDLs 任务

有一个多生产者消费者模式生产者1->消费者1/生产者2->消费者2/生产者3，每个生产者都使用一个完成服务，但是我在编写完成后的逻辑编码时遇到了问题。

问题是 main 将一些 (x) 任务放在生产者 1 队列中，这反过来会导致生产者 1 将 (y) 任务放在生产者 2 上，并且生产者 2 会将 z 任务放在生产者 3 上。x、y 和 z 是不同，因此 main 不能只查看生产者 3 完成队列并获取 z 期货，因为 main 不知道 z。

所以我尝试了毒丸的想法和一个初始化为3的CountDownLatch，主要知道生产者1只有x个任务，所以我可以在最后提交毒丸，然后当生产者1收到这个它可以减少锁存器并发送毒药药丸发送给生产者 2，生产者 2 接收到一个它会减少锁存器，然后将其发送给生产者 3。当生产者 3 收到它时，它会减少锁存器。main 只是做 countdownlatch.await() 并且在所有任务完成之前无法继续。但这只有在每个 executor 服务仅限于一个线程时才有效，因为当生产者收到毒丸时，它仅意味着所有前面的任务都已启动尚未完成。

那么我该如何解决这个问题，我想我一定是在某个地方错过了一个更简单的解决方案。

我有一个 JPanel，它以与JOptionPane.showMessageDialog(...). 但我不使用此选项，因为我想更改大小、消息、按钮位置以及更多内容。但最终的实用程序必须相同，一旦选择就返回所选选项。问题是我不想轮询一个可能被初始化为 null 的属性，直到它从actionPerformed(...)方法中改变。相反，我曾想过以某种方式使用 a CompletionService，但我无法思考如何准确地做到这一点。我该如何定义它？因为我猜Future必须在一个getSelectedOption()方法中选择它，但它必须在一个方法中生成 actionPerformed(...)。这该怎么做？

我有这个代码：

主类：

输出： ...

如果一个线程找到密码并将stopped 设置为true，则另一个线程不会立即停止。为什么？

在我的应用程序中，我必须从主应用程序线程异步处理多个作业并收集每个作业的结果。我有一个简单的 Java 解决方案，它使用 ExecutorService 和 ExecutorCompletionService 来收集作业结果。

现在我想将我的代码转换为 Spring 解决方案。文档向我展示了如何使用 ExecutorService 和 @Async 注释，但我不确定如何以及是否可以收集多个作业的结果。

换句话说：我正在寻找与 CompletionService 等效的 Spring。有这样的事吗？

我当前的代码：

我需要提交几个作业，它们将使用相同的输入文件夹，但在不同的输出文件夹中产生不同的结果。这些作业应该并行运行并且不相互依赖。

是否有任何简单的方法可以等待所有这些作业的完成（例如 Java 并发包中的 CompletionService），或者我需要从头开始创建它 - 记住所有作业的作业 ID 并定期检查所有作业的状态？

我正在尝试一起使用 ExecutorCompletionService 和 ScheduledExecutorService 。

我需要做的是安排不同的活动，每个活动都有“执行前的延迟”，然后根据上次运行的结果“重新安排它们”（不同的延迟）。

我遇到的问题是我不能使用 ExcecutorCompletionService 提交“延迟”

我尝试了以下，但它永远阻塞......

显然，我错过了 Java 语言中的一个基本问题。

反正有没有安排任务到 ScheduledExecutorService 以便 CompletionService “知道它”？

如果我把“延迟在可调用”通过 new ActionOne(delay) 创建可调用；并使用 CompletionService.submit(..) 它可以工作。

所以我想最后一个问题是：ScheduledExecutorService 有什么比 Thread.sleep(delay) 做这件事的方式更好的东西吗？

我需要在多核机器上处理大量（>1 亿）请求（每个请求是处理数据文件中的一行，并且涉及与远程系统的一些 I/O。虽然细节并不重要，具体任务是从一些数据文件中加载分布式Hazelcast地图）。执行将通过ThreadPoolExecutor处理。一个线程将读取文件，然后将数据提交给多个独立线程以将其放入地图中。该机器有 32 个内核，因此有足够的可用于并行加载地图。

由于请求量很大，创建任务并将它们排队到执行器服务的常用方法是不可行的，因为排队的任务会占用过多的内存。

这带来了ExecutorCompletionService。有了它，当先前的操作完成时将提交一个任务，这通过调用take()（或poll()，如果适用）知道。当执行器服务的所有线程都被使用时，这将正常工作。但是，“加载所有线程”还没有完成。有两个阶段：

• 填满队列：当池中仍有未使用的线程时，将任务提交给 ExecutorCompletionService 并且不要等到提交更多

• 喂入队列：一旦线程全部使用完毕，只有在前一个任务完成后才提交任务。因此，行将尽可能快地送入，但不会更快，也不会排队。

上面可以编码，但我想知道上面的逻辑是否已经实现，我不知何故错过了它。我问是因为它看起来是一个常见的场景。