java - ThreadPoolExecutor.shutdownNow() 没有在 Thread 中抛出 InterruptedException

Question

我正在实现一个传输服务器程序，该程序从客户端获取消息(通过控制台输入)，然后将其转发到某种邮箱。

为了允许不同客户端同时接收多个消息，我首先创建了一个实现Runnable 接口(interface)的类。每个此类实例都将处理与一个客户端的通信:

public class ClientConnection implements Runnable {

    //...

    //...

    @Override
    public void run() {
        try {
            // prepare the input reader and output writer
            BufferedReader reader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
            PrintWriter writer = new PrintWriter(clientSocket.getOutputStream(), true);

            Message message = new Message();
            String request = "";

            // read client requests
            while ((request = reader.readLine()) != null) {

                System.out.println("Client sent the following request: " + request);
                String response;
                if (request.trim().equals("quit")) {
                    writer.println("ok bye");
                    return;
                }

                response = message.parseRequest(request);
                if (message.isCompleted()) {
                    messagesQueue.put(message);
                    message = new Message();
                }
                writer.println(response);
            }

        } catch (SocketException e) {
            System.out.println("ClientConnection: SocketException while handling socket: " + e.getMessage());
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        } catch (InterruptedException e) {
            System.out.println("Client Connection was interrupted!");
            e.printStackTrace();
        } finally {
            if (clientSocket != null && !clientSocket.isClosed()) {
                try {
                    clientSocket.close();
                } catch (IOException ignored) {}
            }

        }

    }
}

我确实有一个父线程负责启动和管理所有 ClientConnection runnables:

@Override
public void run() {

    clientConnectionExecutor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
    while (true) {

        Socket clientSocket;

        try {
            // wait for a Client to connect
            clientSocket = serverSocket.accept();

            ClientConnection clientConnection = new ClientConnection(clientSocket, messagesQueue);
            clientConnectionExecutor.execute(clientConnection);

        } catch (IOException e) {
            // when this exception occurs, it means that we want to shut down everything
            clientConnectionExecutor.shutdownNow();  // force terminate all ClientConnections
            return;
        }
    }
}

现在根据this Stackoverflow Question ，我本以为一旦 shutdownNow(); 被调用，InterruptedException 就会在我的 ClientConnection.run() 中抛出方法，在那里，它应该打印 Client Connection was interrupted!。但这并没有发生，因此 catch 子句似乎永远不会到达，输入读取循环一直在继续。

我在另一个 Stackoverflow 问题中读到，这可能与 block 中的某些其他代码行有关，似乎正在消耗 InterruptedException，但没有任何关于什么代码行可以做到这一点的具体信息。所以我很感谢任何提示。

编辑:事实证明，只要我通过在客户端上键入“quit”手动退出循环，循环就会退出，然后打印 Client Connection was interrupted!。因此，只要循环正在运行，异常似乎就会以某种方式被忽略，并且只会在之后处理。

Answer 1

来自 Oracle 文档 shutdownNow :

There are no guarantees beyond best-effort attempts to stop processing actively executing tasks. For example, typical implementations will cancel via Thread.interrupt(), so any task that fails to respond to interrupts may never terminate.

如果您查看 ThreadPoolExecutor资源，你会发现shutdownNow使用此代码中断线程:

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }

你的 ClientConnection不检查标志 Thread.interrupted .由于帖子中的信息，我无法弄清楚哪个方法会抛出 InterruptedException .可能是其他一些方法，例如 readLine读者或作者，阻塞线程，因为他们使用套接字的 InputStream和 OutputStream并且因为很明显，如果数据不是立即可用，套接字的流会阻塞线程。

例如，我写了这段代码来测试它:

class Example {
    public static void main(String[] args) {
        Thread thread = new Thread(() -> {
            try(ServerSocket serverSocket = new ServerSocket()) {
                serverSocket.bind(new InetSocketAddress(8080));
                Socket socket = serverSocket.accept();
                int dataByte = socket.getInputStream().read();
                System.out.println(dataByte);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        thread.start();
        thread.interrupt();
    }
}

在 OpenJdk-16.0.2 上没有实际中断。

我看到您的问题有两种可能的解决方案:

1. 检查 Thread.interrupted在while里面如果您确定 Socket 则循环不会阻塞您的线程。

2. 如果您不确定，请使用 SocketChannel在非阻塞模式下而不是 Socket用于检查 Thread.interrupted手动。

对于第二种方式，我将示例转换为:

class Example {
    public static void main(String[] args) {
        Thread thread = new Thread(() -> {
            try(ServerSocketChannel serverSocket = ServerSocketChannel.open()) {
                serverSocket.configureBlocking(false);
                serverSocket.bind(new InetSocketAddress(8080));

                SocketChannel socket = null;

                while (socket == null) {
                    socket = serverSocket.accept();

                    if (Thread.interrupted()) {
                        throw new InterruptedException();
                    }
                }

                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                socket.read(byteBuffer);
                byte[] bytes = new byte[byteBuffer.limit()];
                byteBuffer.flip();
                byteBuffer.get(bytes);
                System.out.println(new String(bytes, StandardCharsets.UTF_8));
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                System.out.println("Interrupted successfully");
            }
        });
        thread.start();
        thread.interrupt();
    }
}

有效。

祝你使用 Java :)