标签: threadpool

是否存在线程池(在Java中)的任何实现,以确保在同一线程上执行相同逻辑ID的所有任务？

我所追求的逻辑是,如果在给定逻辑ID的特定线程上已经执行了任务,则在同一线程上安排具有相同ID的新任务.如果没有线程为同一ID执行任务,则可以使用任何线程.

这将允许并行执行不相关ID的任务,但是同一ID的任务将以串行和提交的顺序执行.

如果没有,是否有任何关于如何扩展ThreadPoolExecutor以获得此行为的建议(如果可能的话)？

UPDATE

花了更长时间考虑这个问题,我实际上并不要求在同一个线程上执行相同逻辑ID的任务,只是它们不会同时执行.

这方面的一个例子是处理客户订单的系统,可以同时处理多个订单,但不能同一个客户(并且必须按顺序处理同一客户的所有订单).

我现在采用的方法是使用标准的ThreadPoolExecutor,自定义BlockingQueue并Runnable使用自定义包装器进行包装.的Runnable包装器逻辑是:

1. 以原子方式尝试将ID添加到并发"running"set(ConcurrentHashMap)以查看当前是否正在运行相同ID的任务
   • 如果添加失败,请将任务重新推送到队列的前面并立即返回
   • 如果成功,继续
2. 运行任务
3. 从"正在运行"的集合中删除任务的关联ID

然后,队列的poll()方法只返回具有当前不在"运行"集中的ID的任务.

这样做的问题在于,我确信会有很多我没有想过的极端情况,因此需要进行大量的测试.

我正在尝试调试很少有死锁的自定义线程池实现.所以我不能使用像gdb这样的调试器,因为我在遇到死锁之前点击了100次"启动"调试器.

目前,我在shell脚本的无限循环中运行线程池测试,但这意味着我看不到变量等等.我正在尝试std::cout数据,但这会减慢线程并减少死锁的风险,这意味着我可以在获取消息之前等待1小时的无限.然后我没有得到错误,我需要更多的消息,这意味着再等一个小时......

如何有效地调试程序,以便它一遍又一遍地重启直到它死锁？(或者我应该用所有代码打开另一个问题以获得一些帮助？)

先感谢您 !

奖金问题:如何检查一切顺利std::condition_variable？您无法确定哪个线程处于睡眠状态,或者条件是否出现竞争wait条件.

ScheduledThreadPoolExecutor如果没有工作要做,我想有一个也停止最后一个线程,并且如果有新任务则创建(并保持线程活动一段时间).但是,一旦没有更多的工作要做,它应该再次丢弃所有线程.

我天真地创建了它,new ScheduledThreadPoolExecutor(0)但结果是,没有创建任何线程,也没有执行任何计划任务.

任何人都可以告诉我,如果我能够实现我的目标而不用自己的包装ScheduledThreadpoolExecutor吗？

提前致谢!

我有这个代码:

var list = new List<int>();
for(int i=0;i<10;i++) list.Add(i); 
for(int i=0;i<10;i++)
{
     ThreadPool.QueueUserWorkItem(
         new WaitCallback(x => {
             Console.WriteLine(x);  
         }), list[i]);
} 

我想知道所有线程池线程什么时候完成他们的工作.我该怎么做？

自从我想要添加线程时编写ASP.NET应用程序以来,我可以通过3种简单的方法在ASP.NET应用程序中完成线程化:

• 使用System.Threading.ThreadPool.
• 使用自定义委托并调用其BeginInvoke方法.
• 在System.Threading.Thread类的帮助下使用自定义线程.

前两种方法提供了一种快速方法来为您的应用程序启动工作线程.但不幸的是,它们损害了应用程序的整体性能,因为它们使用ASP.NET使用的相同池中的线程来处理HTTP请求.

然后我想使用新的Task或async/await来编写IHttpAsyncHandler.您可以找到的一个例子是Drew Marsh在这里解释的:https://stackoverflow.com/a/6389323/261950

我的猜测是使用Task或async/await仍然消耗来自ASP.NET线程池的线程,我不想要显而易见的原因.

你可以告诉我,如果我可以在后台线程上使用Task(async/await),就像使用System.Threading.Thread类而不是线程池一样吗？

在此先感谢您的帮助.

托马斯

我正在处理的当前Python应用程序需要使用1000多个线程(Pythons线程模块).并不是说任何单个线程都在最大cpu周期工作,这只是我正在创建的Web服务器负载测试应用程序.IE模拟200个firefox客户端,他们都渴望进入Web服务器并下载小型Web组件,基本上模拟人类在几秒钟内操作,而不是微秒.

所以,我正在阅读各种主题,例如"在Linux/Windows上支持python的线程数等等,我看到了很多不同的答案.一位用户表示,所有内存和Linux内核默认只留出8Meg用于线程,如果超过该线程,则线程开始被内核杀死.

有人说这对CPython来说不​​是问题,因为无论如何只有一个线程一次运行(因为GIL)所以我们可以指定一个庞大的线程??? 这有什么实际的真相？

我正在创建一个可以监控100-150个设备的监控应用程序......现在设计一个监控应用程序,我有两种方法: -

1. 为要监视的每个设备创建一个线程,每个线程将ping(使用ICMP)到设备以了解设备是否在线.这些线程将在特定时间间隔(例如60秒)后无限运行以了解其状态.

2. 创建线程池,并为每个设备将任务提交给线程池.任务是简单ping到设备.因此,在当前的设计中,任务将超过线程池中的线程.例如,假设有100个要监视的设备,将有100个要监视的任务,并且线程池将有40个线程来完成这100个任务.当然,运行下一组任务的持续时间将使得完成线程池中的所有挂起任务.

哪种方法会更好？

最近我们的一个生产tomcat服务器没有响应,因为tomcat的繁忙线程高达200.当我们在重新启动之前进行线程转储时,我们在TIMED_WAITING状态下获得了100个线程,就像这3个线程一样:

""http-bio-7007"-exec-241" daemon prio=10 tid=0x00002aaab107b000 nid=0x59df waiting on condition [0x0000000051239000]
java.lang.Thread.State: TIMED_WAITING (parking)
   at sun.misc.Unsafe.park(Native Method)
   - parking to wait for  <0x0000000580d877d0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
   at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:198)
   at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2025)
   at java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:424)
   at org.apache.tomcat.util.threads.TaskQueue.poll(TaskQueue.java:86)
   at org.apache.tomcat.util.threads.TaskQueue.poll(TaskQueue.java:32)
   at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:945)
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
   at java.lang.Thread.run(Thread.java:662)

""http-bio-7007"-exec-237" daemon prio=10 tid=0x00002aaab186e000 nid=0x596d waiting on condition [0x000000004d1f9000]
java.lang.Thread.State: TIMED_WAITING (parking)
   at sun.misc.Unsafe.park(Native Method)
   - parking to wait for  <0x0000000580d877d0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
   at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:198)
   at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2025)
   at java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:424)
   at org.apache.tomcat.util.threads.TaskQueue.poll(TaskQueue.java:86)
   at org.apache.tomcat.util.threads.TaskQueue.poll(TaskQueue.java:32)
   at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:945)
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
   at java.lang.Thread.run(Thread.java:662)

""http-bio-7007"-exec-236" …

我需要在Java中实现一个线程池(java.util.concurrent),其空闲时线程数达到某个最小值,当作业提交到它上面的速度超过完成执行时,它会增长到上限(但绝不会更远) ,当所有作业完成并且不再提交任何作业时,缩小回到下限.

你会如何实现这样的东西？我想这将是一个相当常见的使用场景,但显然java.util.concurrent.Executors工厂方法只能创建固定大小的池和池,当提交许多作业时,这些池和池无限增长.本ThreadPoolExecutor类提供corePoolSize和maximumPoolSize参数,但它的文档似乎暗示,只有这样,才能不断有超过corePoolSize线程在同一时间使用有界作业队列,在这种情况下,如果你已经达到maximumPoolSize线程,你会得到工作拒绝你必须自己处理？我想出了这个:

//pool creation
ExecutorService pool = new ThreadPoolExecutor(minSize, maxSize, 500, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<Runnable>(minSize));
...

//submitting jobs
for (Runnable job : ...) {
    while (true) {
        try {
            pool.submit(job);
            System.out.println("Job " + job + ": submitted");
            break;
        } catch (RejectedExecutionException e) {
            // maxSize jobs executing concurrently atm.; re-submit new job after short wait
            System.out.println("Job " + job + ": rejected...");
            try {
                Thread.sleep(300);
            } catch (InterruptedException …

如何创建scala.concurrent.ExecutionContext？

文档通常给出一个总体摘要,并提到"默认"实现scala.concurrent.ExecutionContext.global.

尽管如此,有时你必须创建你的个人EC,而不使用akka和其他这样的工具.