标签: threadpool

当ASP.NET收到请求时,它如何确定是为其提供服务还是对其进行排队？我问,因为我正在监视服务器上的性能计数器并且CPU没有达到最大值并且有大量可用的工作线程,但我仍然看到多达200个请求排队.

我正在开发一个需要并发的Java守护进程:一个无限循环,它监听作业队列(redis)并将每个作业分配给一个worker.工作人员不一定要返回值.

我发现Executors非常有用,我使用ThreadPoolExecutor来维护许多工作线程.

但是,这些工作者运行的第三方代码需要尽可能隔离,避免共享静态属性.

我的问题:是否有任何Java库/框架提供类似于Executors的功能,例如:

• 工人池
• 自动调整池大小

..但产生进程而不是线程？

至少一个星期以来,我一直在努力解决这个问题(也学到了新的东西--WCF是一个主要的PITA).

这是我的问题:我的应用程序中有一个场景,在某些时候会冻结整个客户端,例如,永远(因为我禁用了超时,因为客户端和服务器都在受控环境中).死锁恰好发生在同一个呼叫中,我认为是由于它之前的请求突发.

检查客户端上的死锁堆栈跟踪给了我:

[In a sleep, wait, or join] 
WindowsBase.dll!System.Windows.Threading.DispatcherSynchronizationContext.Wait(System.IntPtr[] waitHandles, bool waitAll, int millisecondsTimeout) + 0x26 bytes 
mscorlib.dll!System.Threading.SynchronizationContext.InvokeWaitMethodHelper(System.Threading.SynchronizationContext syncContext, System.IntPtr[] waitHandles, bool waitAll, int millisecondsTimeout) + 0x1c bytes   
[Native to Managed Transition]  
[Managed to Native Transition]  
mscorlib.dll!System.Threading.WaitHandle.InternalWaitOne(System.Runtime.InteropServices.SafeHandle waitableSafeHandle, long millisecondsTimeout, bool hasThreadAffinity, bool exitContext) + 0x2b bytes 
mscorlib.dll!System.Threading.WaitHandle.WaitOne(int millisecondsTimeout, bool exitContext) + 0x2d bytes    
mscorlib.dll!System.Threading.WaitHandle.WaitOne() + 0x10 bytes 
System.Runtime.DurableInstancing.dll!System.Runtime.TimeoutHelper.WaitOne(System.Threading.WaitHandle waitHandle, System.TimeSpan timeout) + 0x7c bytes 
System.ServiceModel.dll!System.ServiceModel.Channels.OverlappedContext.WaitForSyncOperation(System.TimeSpan timeout, ref object holder) + 0x40 bytes    
System.ServiceModel.dll!System.ServiceModel.Channels.PipeConnection.WaitForSyncRead(System.TimeSpan timeout, bool traceExceptionsAsErrors) + 0x38 …

有没有办法使用ExecutorService来暂停/恢复特定的线程？

private static ExecutorService threadpool = Executors.newFixedThreadPool(5);

想象一下,我想要停止作为id = 0的线程(假设为每个线程分配一个增量id,直到达到线程池的大小).

过了一会儿,按一个按钮让我们说,我想恢复那个特定的线程并让所有其他线程保持其当前状态,可以暂停或恢复.

我在Java文档中找到了一个未完成的PausableThreadPoolExecutor版本.但它不适合我需要的东西,因为它恢复了池中的所有线程.

如果没有办法使用ExecutorService的默认实现,那么任何人都可以指出这个问题的Java实现吗？

谢谢!

我的问题是:石英调度程序如何工作,它与实现Runnable接口(基本上是一个线程)的普通类有什么不同,它根据指定的时间间隔唤醒并执行所需的工作？

它只是一种方便的处理方式(使用石英调度程序),比如通过XML文件指定作业配置,以及向现有调度程序轻松添加新作业,还是有更多内容？更多,我的意思是它是否进行任何类型的优化,例如它不会在整个持续时间内挂起并释放它？它是一种轮询类型的机制,线程保持轮询系统时间并查看指定的时间间隔是否已经过,或者它是否使用系统时钟进行某种注册,时钟本身通知石英调度程序？

如果上述问题需要进一步澄清,请与我们联系.

我们正在开发一个应用程序,通过接收来自3个不同来源的消息,可以影响一组对象.每条消息(来自任何来源)都有一个对象作为其目标.每个消息接收器将在其自己的线程上运行.

我们希望消息的处理(在接收之后)尽可能高速,因此针对目标对象的消息处理将使用来自线程池的另一个线程来完成.消息的处理将比从发送者读取/接收消息花费更长的时间.

我认为如果池中的每个线程仅专用于特定的一组对象,它会更快,例如:

Thread1 -> objects named A-L
Thread2 -> objects named M-Z

每组对象(或线程)都有一个专用的消息队列待处理.

我的假设是,如果所需的唯一线程同步是在每个接收线程和一个处理线程之间,在需要将消息放入阻塞队列的持续时间内,它将比随机分配工作线程以处理更快消息(在这种情况下,可能有2个不同的线程与同一对象的消息).

我的问题实际上是两个部分:

1. 人们是否同意将工作线程专用于特定对象集合是更好/更快的方法？

2. 假设这是一种更好的方法,现有的Java ThreadPool类是否有办法支持这种方法？或者它是否要求我们编写自己的ThreadPool实现？

感谢您提供的任何建议.

在这个博客中,我找到了一个非常巧妙的例子,说明如何使用boost :: asio创建一个简单的线程池.我基本上想要像这样使用它:

#include <thread>
#include <functional>
#include <boost/asio.hpp>

int main ( int argc, char* argv[] ) {
    asio::io_service io_service;
    asio::io_service::work work(io_service);

    std::vector<std::thread> threadPool;

    for(size_t t = 0; t < std::thread::hardware_concurrency(); t++){
        threadPool.push_back(thread(std::bind(&asio::io_service::run, &io_service)));
    }

    io_service.post(std::bind(an_expensive_calculation, 42));
    io_service.post(std::bind(a_long_running_task, 123));

    //Do some things with the main thread

    io_service.stop();
    for(std::thread& t : threadPool) {
        t.join();
    }
}

据我所知,Boost :: asio主要用于网络IO.但是,我主要想将它用于通用功能.将使用并发问题asio::io_service::strand.

所以我的问题:创建这样的线程池是一个好主意,即使我的程序不使用网络IO？与其他线程池实现相比,是否存在明显的性能损失？如果是这样,那么还有更好的实现吗？

在我们的Delphi XE4应用程序中,我们使用具有MaxExecuting = 4的OmniThreadPool来提高某个计算的效率.不幸的是,我们遇到了间歇性访问违规问题(例如,参见下面的MadExcept错误报告http://ec2-72-44-42-247.compute-1.amazonaws.com/BugReport.txt).我能够构建以下示例来演示该问题.运行以下控制台应用程序后,System.SyncObjs.TCriticalSection.Acquire中的访问冲突通常会在一分钟左右内发生.任何人都可以告诉我在下面的代码中我做错了什么,或者告诉我另一种方法来实现预期的结果？

program OmniPoolCrashTest;

{$APPTYPE CONSOLE}

uses
  Winapi.Windows, System.SysUtils,
  DSiWin32, GpLists,
  OtlSync, OtlThreadPool, OtlTaskControl, OtlComm, OtlTask;

const
  cTimeToWaitForException = 10 * 60 * 1000;  // program exits if no exception after 10 minutes
  MSG_CALLEE_FINISHED = 113; // our custom Omni message ID
  cMaxAllowedParallelCallees = 4; // enforced via thread pool
  cCalleeDuration = 10; // 10 miliseconds
  cCallerRepetitionInterval = 200; // 200 milliseconds
  cDefaultNumberOfCallers = 10; // 10 callers each issuing 1 call every 200 milliseconds

var …

我正在对Spring应用程序进行一些负载测试,现在我对它的配置有点困惑ThreadPoolTaskExecutor.

内部使用的文档ThreadPoolExecutor描述corePoolSize为"要保留在池中的线​​程数,即使它们是空闲的,[...]"和maximumPoolSize"池中允许的最大线程数".

这显然意味着maximumPoolSize限制池中线程的数量.但相反,限制似乎是由corePoolSize.实际上我配置只是corePoolSize与100一个让maximumPoolSize未配置(这意味着使用默认值:Integer.MAX_VALUE= 2147483647).

当我运行负载测试,我可以看到(通过查看日志),为所进行的工作线程从编号worker-1来worker-100.所以在这种情况下,线程池大小受限于corePoolSize.即使我设置maximumPoolSize为200或300,结果也完全一样.

为什么价值对maximumPoolSize我的情况没有影响？

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(100);
    taskExecutor.setThreadNamePrefix("worker-");
    return taskExecutor;
}

解

我在文档中找到了解决方案:" 如果运行的corePoolSize不止于maxPoolSize,但只有队列已满,才会创建新线程 ".默认队列大小是Integer.MAX_VALUE.如果我限制队列,一切正常.

我正在尝试用 Rust 制作一个多线程 tcp 通信程序

这个想法是主线程上存在一个侦听套接字，并且当连接进入时，工作由工作线程处理

我之前使用了 Rust 书中找到的 ThreadPool 方法，但据我了解，tokio 能够“自动”将工作分配给池中的线程

我对操作系统线程和 tokio 任务之间的区别感到困惑（主要是因为您用来spawn创建两者）

这是一些代码

fn main() {
    println!("Hello World!");
    let socket = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 103, 7)), 2048);
    println!("socket -> {}", socket);

    // You need to use the tokio runtime to execute async functions
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        let listener = StartListen::new(socket).await.unwrap();
    });
}

我StartListen在另一个文件中定义了

// Defines the StartListen class
pub struct StartListen {
    listener: TcpListener,
}


// Implementation for …