- Java 双重比较
- java - 比较器与 Apache BeanComparator
- Objective-C 完成 block 导致额外的方法调用?
- database - RESTful URI 是否应该公开数据库主键?
我正在尝试使用 ThreadPoolExecutor 来安排任务,但在其策略方面遇到了一些问题。这是其声明的行为:
我想要的行为是这样的:
基本上我不希望任何任务被拒绝;我希望他们在无界队列中排队。但我确实希望拥有最多 maximumPoolSize 线程。如果我使用无界队列,它在达到 coreSize 后永远不会生成线程。如果我使用有界队列,它会拒绝任务。有什么办法解决这个问题吗?
我现在考虑的是在 SynchronousQueue 上运行 ThreadPoolExecutor,但不是直接向它提供任务 - 而是将它们提供给一个单独的无界 LinkedBlockingQueue。然后另一个线程从 LinkedBlockingQueue 进入 Executor,如果一个线程被拒绝,它会再次尝试直到没有被拒绝。不过,这似乎很痛苦,而且有点麻烦 - 有没有更简洁的方法来做到这一点?
最佳答案
可能没有必要根据请求对线程池进行微观管理。
缓存线程池将重新使用空闲线程,同时还允许潜在的无限并发线程。这当然会导致突发期间上下文切换开销导致性能失控。
Executors.newCachedThreadPool();
更好的选择是对线程总数进行限制,同时放弃确保首先使用空闲线程的概念。配置更改为:
corePoolSize = maximumPoolSize = N;
allowCoreThreadTimeOut(true);
setKeepAliveTime(aReasonableTimeDuration, TimeUnit.SECONDS);
根据这种情况推理,如果执行程序的线程少于 corePoolSize
,那么它一定不会很忙。如果系统不是很忙,那么启动一个新线程也没什么坏处。这样做会导致您的 ThreadPoolExecutor
在低于允许的最大工作人员数量时始终创建一个新工作人员。只有当最大数量的worker在“运行”时,才会给空闲等待任务的worker分配任务。如果工作人员在没有任务的情况下等待 aReasonableTimeDuration
,则允许终止。对池大小使用合理的限制(毕竟,只有这么多 CPU)和相当大的超时(以防止线程不必要地终止),可能会看到预期的好处。
最后一个选项是骇人听闻的。基本上,ThreadPoolExecutor
在内部使用 BlockingQueue.offer
来确定队列是否有容量。 BlockingQueue
的自定义实现总是可以拒绝 offer
尝试。当 ThreadPoolExecutor
无法向队列提供
任务时,它将尝试创建一个新的 worker。如果无法创建新的 worker,则会调用 RejectedExecutionHandler
。此时,自定义 RejectedExecutionHandler
可以强制将 put
放入自定义 BlockingQueue
。
/** Hackish BlockingQueue Implementation tightly coupled to ThreadPoolexecutor implementation details. */
class ThreadPoolHackyBlockingQueue<T> implements BlockingQueue<T>, RejectedExecutionHandler {
BlockingQueue<T> delegate;
public boolean offer(T item) {
return false;
}
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
delegate.put(r);
}
//.... delegate methods
}
关于java - ThreadPoolExecutor 政策,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/3419380/
这是正确的吗? ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor)
我只是对我编写的一些代码感到非常困惑。我惊讶地发现: with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
我正在尝试同时使用 InheritableThreadLocal 和 ThreadPoolExecutor。 这是因为 ThreadPoolExecutor 为每个池重用线程(毕竟它是一个池),这意味
concurrent.futures.ThreadPoolExecutor 通过 传递 function 到执行器 executor.submit(my_function) 像这样: def my_f
将一个简单的测试程序放在一起,应该并行执行一些任务。每次我们提交6个任务,等待完成。然后,又提交了一组任务。 import java.util.concurrent.*; public class
我正在运行一段 python 代码,其中多个线程通过线程池执行程序运行。每个线程都应该执行一项任务(例如获取网页)。我想要做的是终止所有线程,即使其中一个线程失败。例如: with ThreadPoo
我有一个使用阻塞队列的ThreadPoolExecutor,并且正在尝试调试一个问题,在该问题中,我怀疑任务在ThreadPoolExecutor的队列中停留的时间太长,无法执行。我正在尝试验证这一理
我正在尝试使用 futures backport 包在 Python 中使用 ThreadPoolExecutor。然而,问题是所有线程都是同时执行的,所以没有实际的池化发生。更具体地说,我得到了该函
我有两个 list : a = [1, 2, 3, 4] b = [9, 8, 7, 6] 我希望将这两个列表的每个组合作为参数传递给我正在执行多线程处理的函数: def test(hello, wo
当我们谈论ThreadPoolExecutor时,核心池大小和最大池大小之间到底有什么区别? 可以用例子来解释吗? 最佳答案 来自this blog post : Take this example.
我对“concurrent.futures”的并行处理相当陌生,我正在测试一些简单的实验。我编写的代码似乎可以工作,但我不确定如何存储结果。我尝试创建一个列表(“ future ”)并将结果附加到该列
我审查了多线程,并尝试实现一个创建单独线程来运行收集进程的应用程序。该过程中的主要方法需要一个变量数组列表,我正在尝试找出一种将数组列表传递给每个线程的方法。 ApplicationContext c
很难说出这里问的是什么。这个问题是含糊的、模糊的、不完整的、过于宽泛的或修辞性的,无法以目前的形式得到合理的回答。如需帮助澄清此问题以便重新打开它,visit the help center 。 已关
我需要读取一个大的 csv 文件(328 MB)并对其进行处理。每行的处理还包括调用 Web 服务。 我是第一次使用ThreadPoolExecutor。我的逻辑是,我将从 csv 中每 100 行吐
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED def div_zero(x): print('I
我有一个带有 run 方法的类,该类的 main 方法中的计时器正在使用以下代码调用该类: Timer timer = new Timer(); timer.scheduleAtFixedRate(n
尝试调试竞争条件,其中我们的应用程序的轮询器线程之一永远不会返回,导致 future 的轮询器永远不会被调度。用抽象术语来说,在捕获问题时隐藏我们的业务逻辑,这就是我们的代码路径。 我们必须更新远程服
下面的错误是什么意思?我怎样才能恢复它? Exception in thread "UserActionProcessor-8811" java.util.concurrent.RejectedExe
我正在致力于增强现有的 Java 应用程序。该应用程序是一个消息处理器,每天处理数百万条消息。它基本上是使用 Core Java 编写的,线程和队列是使用 Collection 类实现的。 在此应用程
我想更新使用用户定义线程池的旧代码。我想使用 java ThreadPoolExecutor,但问题是发送到线程池的请求不是可运行的。有什么方法可以将 ThreadPoolExecutor 与现有请求
我是一名优秀的程序员,十分优秀!