标签: threadpool

我一直在使用ThreadPoolExecutor和JDK6来解决线程池的不同策略.我有一个优先级队列工作,但不知道我是否喜欢keepAliveTime之后池的大小(你得到的是无限队列).所以,我正在使用LinkedBlockingQueue和CallerRuns策略查看ThreadPoolExecutor.

我现在遇到的问题是,池正在升级,因为文档解释了它应该,但是在任务完成并且keepAliveTime发挥作用后,getPoolSize显示池减少到零.下面的示例代码可以让您看到我的问题的基础:

public class ThreadPoolingDemo {
    private final static Logger LOGGER =
         Logger.getLogger(ThreadPoolingDemo.class.getName());

    public static void main(String[] args) throws Exception {
        LOGGER.info("MAIN THREAD:starting");
        runCallerTestPlain();   
    }

    private static void runCallerTestPlain() throws InterruptedException {
        //10 core threads, 
        //50 max pool size, 
        //100 tasks in queue, 
        //at max pool and full queue - caller runs task
        ThreadPoolExecutor tpe = new ThreadPoolExecutor(10, 50,
            5L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(100),
            new ThreadPoolExecutor.CallerRunsPolicy());

        //dump 5000 tasks on the queue
        for (int i = 0; i < 5000; i++) …

我有一个类,说"MyComputation",它在一个长构造函数中进行了大量的计算.它自己执行时通常需要大约20ms才能运行(没有磁盘i/o或网络操作).这个类的100个左右的实例是由父类创建的,比如"ComputeParent",它将它们作为工作项在ThreadPool中排队:

ThreadPool.QueueUserWorkItem(myComputationCall, my_computation_data);

"myComputationCall"看起来像这样:

    public static void myComputationCall(Object my_computation_data)
    {
        try
        {
            MyDataObject data = (MyDataObject)my_computation_data;

            var computation_run = new MyComputation(data.parameter1, data.parameter2);

            data.result = computation_run.result;
        }
        finally
        {
            if (Interlocked.Decrement(ref num_work_items_remaining) == 0)
                done_event.Set();
        }
    }

done_event是一个静态的ManualResetEvent:

    private static ManualResetEvent done_event;

    ...

    done_event = new ManualResetEvent(false);

对于各种输入参数,我运行ComputeParent约500次左右.所以我有很多嵌套类.问题是执行ComputeParent所需的时间逐渐增加.在运行每个特定ComputeParent所需的时间之间会有一定的变化,但是时间量会相当稳定地增加(几何上,每次连续迭代需要更长的时间).

程序的内存消耗虽然很高(~300MB),但并不会随着时间的推移而显着增加.它运行在具有8个逻辑核心的计算机上,处理器的使用似乎非常突发.我不确定还有什么可能与这个问题有关.

我不希望不必通过批处理文件运行ComputeParent,尽管在完成此操作后似乎不会出现问题.

我正在阅读一篇关于HttpContext和CallContext的文章,看看线程敏捷性.这是什么意思？

我已经阅读(和使用过)async/await一段时间了很长一段时间但我还有一个问题我无法得到答案.说我有这个代码.

private async void workAsyncBtn_Click(object sender, EventArgs e)
{
    var myTask = _asyncAwaitExcamples.DoHeavyWorkAsync(5);
    await myTask;
    statusTextBox.Text += "\r\n DoHeavyWorkAsync message";
}

它从UI线程调用并返回到UI线程.因此,我可以在此方法中执行UI特定的操作await myTask.如果我曾经使用过,.ConfigureAwait(false)我会在做的时候得到一个线程异常,statusTextBox.Text += "\r\n DoHeavyWorkAsync message";因为我可以通过telled myTask来接受来自线程池的任何可用线程.

我的问题.据我所知,在这种情况下我永远不会离开UI线程,它仍然是异步运行,UI仍然是响应式的,我可以同时启动多个任务,从而加快我的应用程序.如果我们只使用一个线程,它怎么能工作？

谢谢!

编辑Sievajet

private async void workAsyncBtn_Click(object sender, EventArgs e)
{
    await DoAsync();
}

private async Task DoAsync()
{
    await Task.Delay(200);
    statusTextBox.Text += "Call to form";
    await Task.Delay(200);
}

你能想出为什么这段代码不起作用并且总是输出“完成”的任何原因，但第二个例子没有任何问题。我正在使用最新的 JDK (8u45)。

公共静态类 MyClass 实现了 Runnable {

        @覆盖
        公共无效运行（）{
            尝试 {
                线程睡眠（2000）；
            } catch (InterruptedException ex) {
                System.out.println("中断");
                返回;
            }
            System.out.println("完成");
        }

        公共静态无效主（字符串 [] args）{
             // 找出差异 ->
             ExecutorService executorService = Executors.newWorkStealingPool(4);
             未来未来 = executorService.submit(new MyClass());
             线程睡眠（100）；
             未来.取消（真）；
        }
    }

以下示例完美运行：

公共静态类 MyClass 实现了 Runnable {

        @覆盖
        公共无效运行（）{
            尝试 {
                线程睡眠（2000）；
            } catch (InterruptedException ex) {
                System.out.println("中断");
                返回;
            }
            System.out.println("完成");
        }

        公共静态无效主（字符串 [] args）{
             ExecutorService executorService = Executors.newSingleThreadExecutor();
             未来未来 = executorService.submit(new MyClass());
             线程睡眠（100）；
             未来.取消（真）；
        }
    }

编辑：添加返回和更新的睡眠时间以及另一个示例。

我正在使用 Pool 多线程我的程序，使用 starmap 来传递参数。

我被卡住了，因为我似乎无法找到一种方法来传递 kwargs 以及我在 starmap 函数中传递的zip数组。

pool = Pool(NO_OF_PROCESSES)
branches = pool.starmap(fetch_api, zip(repeat(project_name), api_extensions))

分支请求不完整，因为我仍然无法弄清楚如何传递关键字参数。

def fetch_api(project_name, api_extension, payload={}, headers={}, API_LINK=API_LINK, key=False):
    headers[AUTH_STRING] = 'Gogo'
    call_api = API_LINK + project_name + api_extension
    response_api = requests.get(call_api, headers=headers, params=payload)

    if key: return project_name + ':' + response_api
    else: return response_api

从分支线调用fetch_api() 时，我想将有效负载作为 {'a':1} 和 key=True 传递。

请指导我的方向或答案。谢谢。使用 Python 3.3+。

假设我们有一个线程池数量有限的线程。

Executor executor = Executors.newFixedThreadPool(3);

现在假设其中一项活动任务必须休眠 3 秒（无论出于何种原因）。

executor.execute(() -> {
    try {
        Thread.sleep(3000L);
    } catch (InterruptedException ignore) {}
});

我们如何实现这样一个线程池，当一个任务休眠（或等待监视器/条件）时，线程¹可以有效地用于运行另一个任务？

¹ 通过螺纹我的意思不是“物理” Java线程，因为当线程处于睡眠状态，这将是不可能的。我的意思是，线程池有一个抽象的实现，它实际上似乎允许一个线程在睡眠期间运行另一个任务。关键是总是有 N 个同时运行（非休眠）的任务。

有点类似于监视器处理对关键区域的访问的方式：

• 如果一个线程等待一个资源，该资源可以被另一个线程使用。
• 如果线程得到通知，它就会被放入等待集中以（重新）获得对该资源的访问权。

有没有办法在两个工作线程之间创建直接通信通道new MessageChannel？例如：P我使用worker_threadAPI 创建了一个主线程，它创建了两个工作线程W1并W2

P -> W1
  -> W2

我想直接启用W1和之间的通信W2，而不是通过P使用parentPort.

我通常EF Core在我的 Web 应用程序中使用许多异步方法，如下所示：

await db.Parents.FirstOrDefaultAsync(p => p.Id == id);

我们知道，ThreadPool默认情况下，初始线程数仅限于 CPU 逻辑内核数。用户请求也由ThreadPool.

由于应用程序中有许多异步调用，我是否应该担心处理用户请求或性能问题？

我正在尝试用 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 …