问题标签 [executorservice]

我试图运行ExecutorService对象，FixedThreadPool但遇到了问题。

我希望程序能在纳秒内运行，但它被挂起。我发现我需要Semaphore和它一起使用，这样队列中的项目就不会被累加。

有什么方法可以让我知道池中的所有线程都已使用。

基本代码...

这段代码被挂起，我真的不知道为什么。如何知道池当前是否正在等待所有线程完成？

我有一个必须处理消息数量的要求。这些消息将被另一个进程插入到数据库表中。我所要做的就是检查数据库中的新消息并将其发送给相应的客户电子邮件或http，具体取决于他们的配置。在任何给定时间，消息的数量可能是数千条，而客户数量约为 1000 人。

我打算按照生产者消费者的工作方式来设计这种方式。就像生产者线程轮询数据库中的新消息并将它们放入队列中一样，工作线程读取这些消息和进程。

起初看起来 JMS 是满足此要求的正确解决方案。但是我正在寻找是否有更好的选择，例如 ExecutorService，在以下情况下使用适合此要求的线程池。

1. 如果一条消息传递失败，我将不得不重试几次，至少 24 小时。
2. 如果客户的一条消息传递失败，那么其他消息传递也将失败。因此，在为该客户处理下一条消息之前，我应该尝试发送第一条消息。

这意味着如果我为所有客户的所有消息设置一个队列，那么如果一条消息失败，其他消息将不会被处理。

有人可以就我如何最好地处理这个问题给我建议吗？

提前致谢。

我正在使用该Executor框架Executors.newCachedThreadPool();
，我有一个Runnables 列表，例如 100。
前 50 个，每个创建一个值（存储在一个列表中）供最后 50 个使用。
我想如果我按顺序传递Runnablesexecutor.execute()它们在列表中，它们也将以相同的顺序执行。
但这并没有发生。
这些任务似乎是按随机顺序执行的，它们是交错的，而不是按顺序执行的。
这是它应该如何工作的吗？有什么办法可以解决这个问题？

谢谢

我正在编写一个带有集成服务器的客户端，该服务器应该无限期地等待新连接 - 并在线程上处理每个连接。

我想在主线程上的系统范围内可用的消息处理程序中处理接收到的字节数组。然而，目前处理显然是在客户端线程上完成的。

我查看了 Futures，ExecutorService 的 submit()，但是当我在服务器中创建客户端连接时，数据将返回到服务器线程。如何从那里将它返回到主线程（可能在同步的数据包存储中？）以在不阻塞服务器的情况下处理它？

我当前的实现如下所示：

Update: The robust way seems to be what TomTom suggested - using the newer java.nio instead. As this project is of limited use and more of an experiment, I'd like to know the best way using it with java.io/java.net :)

我正在尝试编写一个解决方案，其中单个线程产生可以并行执行的 I/O 密集型任务。每个任务都有重要的内存数据。所以我希望能够限制当前待处理的任务数量。

如果我这样创建 ThreadPoolExecutor：

然后当队列填满并且所有线程都已经忙时executor.submit(callable)抛出。RejectedExecutionException

executor.submit(callable)当队列已满且所有线程都忙时，我该怎么做才能阻塞？

编辑：我试过这个：

它在某种程度上达到了我想要达到的效果，但是以一种不优雅的方式（基本上被拒绝的线程在调用线程中运行，所以这会阻止调用线程提交更多）。

编辑：（提出问题5年后）

对于阅读此问题及其答案的任何人，请不要将已接受的答案视为一种正确的解决方案。请通读所有答案和评论。

问题

我正在通过ExecutorService运行对某些外部方法的多次调用。我希望能够中断这些方法，但不幸的是它们自己不检查中断标志。有什么办法可以强制从这些方法中引发异常？

我知道从任意位置抛出异常是有潜在危险的，在我的具体情况下，我愿意抓住这个机会并准备好应对后果。

细节

“外部方法”是指来自外部库的一些方法，我不能修改它的代码（我可以，但是每当发布新版本时，它都会成为维护的噩梦）。

外部方法计算量大，不受 IO 限制，因此它们不响应常规中断，我无法强制关闭通道或套接字或其他东西。正如我之前提到的，它们也不检查中断标志。

该代码在概念上类似于：

我试过的

Thread.stop()理论上会做我想做的事，但它不仅已被弃用，而且仅适用于实际线程，而我正在处理执行器任务（它也可能与未来的任务共享线程，例如在线程池中工作时） . 尽管如此，如果没有找到更好的解决方案，我会将我的代码转换为使用老式线程并使用这种方法。

我尝试过的另一个选项是myMethod()使用特殊的“可中断”注释标记和类似的方法，然后使用 AspectJ（我承认我是新手）在那里捕获所有方法调用 - 例如：

但withincode不是递归匹配方法调用的方法，所以我必须将此注释编辑到外部代码中。

最后，这与我之前的问题类似——尽管一个显着的区别是现在我正在处理一个外部库。

考虑我使用 ScheduledExecutorService 为定期执行安排了一个 Runnable，并且发生了一些系统错误，例如 OutOfMemory。它会被无声地吞噬。

正常吗？

为什么它不传播到容器？

处理此类错误的正确方法是什么？

谢谢！

我正在使用 ScheduledThreadPoolExecutor.schedule(Runnable,int,TimeUnit) 来安排实现 Runnable 的类的一些对象。

在某个时间点，我的应用程序正在关闭，我使用ScheduledThreadPoolExecutor.shutdownNow()。根据文档，它返回一个 ScheduledFuture 的列表。

我真正想做的是获取我最初计划的对象，并从中获取一些数据，然后我将输出它说它无法执行。稍后，应用程序将使用它来尝试在应用程序重新启动时执行它。

我有几个线程运行几乎无限的时间和迭代次数。当找到最佳解决方案时，迭代计数将重置为 0。设置最大迭代次数以防止无限循环。

当所有线程都达到最大迭代次数时，我使用 countdownlatch 停止进程。换句话说，当一个线程达到最大迭代次数时，它使用notifyThreadStop()通知我的主线程，当所有线程停止时，触发 countdown()。

注意：我的线程在 FixedThreadPool ExecutorService 中运行。

我想添加一个 maxTime 锁存器。所以我做的是以下

现在，我知道如果倒计时触发了闩锁，这意味着所有线程都已停止，因此我可以关闭我的 ExecutorService。

当达到最大时间时，情况并非如此。所以在（1）中，我想遍历我所有的跑步者，以文明的方式终止他们:-)。为此，我定义了一个函数 requestTermination()，简单地说，就是在我的 runnables 中将迭代计数器设置为 MaxIterationCount。

所以（1）会变成

现在，我需要再次等待，直到所有线程都真正停止，然后我才能继续......嗯，我只是想我可以有一个额外的锁存器并使用它。

所以（2）会变成

当然，我的函数notifyThreadStop()需要对其进行修改，并且需要一个标志告诉它在 mylatch2 上而不是 mylatch 上执行 countdown()。

我想我刚刚回答了我的问题，但是既然所有这些都已经写完了，我将把它留在这里供其他人参考。

现在的问题是：有更好的方法来处理这个问题吗？(1) 或 (2) 中的 shutdownNow() 是唯一需要的吗？知道我的线程必须在退出之前关闭它们自己的日志文件并关闭它们内部的 Callable thread*ss*。

在 ScheduledExecutorService 中运行时，是否有一种很好的方法可以从任务本身中停止重复任务？

可以说，我有以下任务：

从外面看，很容易通过 f.cancel() 取消，但是我怎样才能在指定的地方停止重复呢？（通过 AtomicReference 传递 Future 是不安全的，因为当 scheduleAtFixedRate 返回 f 并且变量设置也很晚时存在一个潜在的窗口，并且任务本身可能已经运行，在引用中看到 null。）