ThreadPoolExecutor 只运行第一个可运行对象

Question

我有一个线程池执行器

private ExecutorService exec=Executors.newCachedThreadPool();

我有一堂课

class MyCallable implements Callable<String>{

    private ReentrantLock lock=new ReentrantLock();

    private Condition cond=lock.newCondition();


    public String call(){

    System.out.println("TEST!!);
    try{
      lock.lock();
      cond.await();
    }finally{
       lock.unlock();
    }
   }
}

我愿意：

for(int i=0;i<10000;i++){
   exec.submit(new MyCallable());
}

我想拥有 10000 个等待调用的线程，但我只能看到一个 TEST！在我的日志中，所以它只提交一次我的任务，为什么它会冻结？？？我的可调用对象在每个对象实例中都有自己的锁，如何有 1000 个等待线程？

Answer 1

我获取了您的代码，填写了缺失的位，修复了编译错误并运行了它。它打印“测试！！”很多次。如果我将任务数量减少到合理的数量，它就会运行完成。如果我将其设置为 10,000，则测试将失败，并显示 OOME 表示无法分配更多线程堆栈。

---

我想拥有 10000 个等待调用的线程

我的测试表明，许多线程会导致您的内存不足。这真的，真的不是一个好主意。

---

这是我使用的确切代码：

package test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorTest {

    public static void main(String[] args) {
        ExecutorService exec = Executors.newCachedThreadPool();
        for (int i = 0; i < 10000; i++) {
            exec.submit(new MyCallable());
        }
    }
}


package test;

import java.util.concurrent.Callable;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

class MyCallable implements Callable<String> {

    private ReentrantLock lock = new ReentrantLock();

    private Condition cond = lock.newCondition();

    public String call() {

        System.out.println("TEST!!");
        try {
            lock.lock();
            cond.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return "foo";
    }
}

---

这是while(true) {await} 测试：

[steve@newbox ~]$ cat ./workspace/test/src/test/CondAwait.java 
package test;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class CondAwait {

    public static void main(String[] args) throws InterruptedException {
         ReentrantLock lock = new ReentrantLock();

         Condition cond = lock.newCondition();
         lock.lock();
         while(true){
             System.out.println("Hi!");
             cond.await();
         }
    }

}
[steve@newbox ~]$ java -cp ./workspace/test/bin test.CondAwait 
Hi!
^C[steve@newbox ~]$ 

简而言之，主线程在第一个await 调用中冻结...正如我所预料的那样。

---

我同意这不是一个好主意，但是那些线程就像一个请求缓存，用户发送一个请求，它的处理器应该被放入线程池中，如果我们每秒有 10k 个请求，你会看到池中有 10k 个线程。请建议如何做得更好。

使用有界线程池，并将边界设置为大致等于您拥有的核心数。

当然，您不会同时暂停 10,000 个可调用对象，但这是一件好事。

如果这些可调用对象旨在模拟等待几秒钟以等待外部事物响应的请求，那么最多 100 个线程池可能是合理的。但是，如果您真的非常需要大规模并行性，那么您应该考虑使用 NIO 选择器，这样少量工作线程就可以运行大量请求，将它们交错而不是阻塞在 I/O 中。

Answer 2

查看java.util.concurrent.CyclicBarrier 课程。正如其 JavaDoc 中所述：

允许一组线程全部等待的同步辅助 彼此到达共同的障碍点。

你的每个线程都需要调用屏障的方法await。这意味着线程被挂起，直到 Barrier 达到预期的线程数或发生超时。

Answer 3

我刚试过你的程序，它对我来说很好用。我可以看到多个任务正在运行。

public class MyCallable implements Callable<String> {

    private ReentrantLock lock = new ReentrantLock();
    private Condition cond = lock.newCondition();

    public String call() {
        System.out.println("TEST!!");
        try {
           lock.lock();
           cond.await();
        } catch (InterruptedException e) {
           e.printStackTrace();
        } finally {
           lock.unlock();
        }
        return "done";
    }

    public static void main(String[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
           service.submit(new MyCallable());
        }
    }
}

我建议 10000 个并发任务不是一个好主意。你有多少个处理器？ ;)