无法通过线程池创建更多线程来处理套接字（块 IO）

Question

我想创建一个新的thead并将其分配给线程池执行器，以便可以同时处理它。这种线程池的核心大小为 5，最大为 20。但是在下面的代码中，它不能创建比核心大小更多的线程。如果我取消注释代码if (socket != null) { socket.close(); }（实际上，我不能在这里关闭它，因为套接字应该在TestHandler中处理），可以创建更多线程，最多可以创建20个线程池。 那么为什么会发生这种情况呢？有没有人可以帮忙解释一下？ workQueue 的大小可以相应调整，我使用apache ab 进行测试。例如ab -n 1000 -c 10 http://localhost:8080/

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;
import java.util.concurrent.*;

public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        MultiThreadServer server = new MultiThreadServer();
        server.start();
    }

    private static ThreadFactory builder = new ThreadFactoryBuilder().setNameFormat("Demo Task Executor #%d").build();

    public void start() throws IOException {
        ServerSocket serverSocket = new ServerSocket(8080);
        Socket socket = null;
        ExecutorService executor = new ThreadPoolExecutor(5,
                20,
                10,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(10),
                builder);
        for (; ; ) {
            socket = serverSocket.accept();

            executor.execute(new TestHandler(socket));
//            if (socket != null) {
//                socket.close();
//            }
        }
    }

    class TestHandler implements Runnable {
        private Socket socket;

        public TestHandler(Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run() {
            try {
                Thread.sleep(new Random().nextInt(100));
                socket.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread());
        }
    }

}

Answer 1

如果您检查ExecutorService 类中execute 方法的实现，您可以看到以下代码：

//...................

        if (isRunning(c) && workQueue.offer(command)) {
//...................
        }
        else if (!addWorker(command, false))
            reject(command);

//...................

所以只有当workQueue 已满时才会创建一个新线程。您已使用new ArrayBlockingQueue<>(10) 初始化您的池。这意味着不会为前 15 个并发请求（corePoolSize + 队列容量）创建新线程。

要查看例如 6 个线程，只需尝试运行 ab -n 1000 -c 16 http://localhost:8080/ 或减少队列初始容量。但请注意，当队列已满且线程数等于maximumPullSize 时，任何新请求都将被拒绝。