通过消息 Java 关闭多线程套接字服务器

Question

我有一个多线程进程来监听消息并处理它们。如果收到的消息之一是“关闭”，我希望能够关闭进程。我已经实现了除关机部分之外的所有内容。

我有一个使用start方法扩展java.net.ServerSocket 的“Multi”类。里面……

java.net.Socket socket = null;
while (true) {
            try {
                socket = this.accept();
                new Thread(new SocketThread(socket, verifier, threading)).start();

            } catch (IOException e) {
                e.printStackTrace();
            }
        }

那个 SocketThread 是 implements Runnable 的另一个类。有什么办法可以使这项工作？

Answer 1

我以前不必这样做（因为我通常发现自己不必编写太多原始 SocketServer），但是通过这种类型的线程协调，我会尝试以下类似的操作...

public class App {
    public static void main(String[] args) {
        new App().run();
    }

    public void run() {
        try {
            System.out.println("Starting...");
            AtomicBoolean running = new AtomicBoolean(true);
            Collection<Socket> sockets = new ArrayList<>();
            Collection<Thread> threads = new ArrayList<>();
            try (ServerSocket socketServer = new ServerSocket(10101)) {
                System.out.println("Started.");
                while (running.get()) {
                    Socket socket = socketServer.accept();
                    sockets.add(socket);
                    if (running.get()) {
                        Thread thread = new Thread(new SocketHandler(socket, running));
                        thread.start();
                        threads.add(thread);
                    }
                }
                System.out.println("Stopping...");
                sockets.forEach(socket -> {
                    try {
                        socket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
                threads.forEach(thread -> {
                    try {
                        thread.join();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }
            System.out.println("Stopped.");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    static class SocketHandler implements Runnable {
        private final Socket socket;
        private final AtomicBoolean running;

        SocketHandler(Socket socket, AtomicBoolean running) {
            this.socket = socket;
            this.running = running;
        }

        @Override
        public void run() {
            try {
                System.out.println("Client connected.");
                try (BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()))) {
                    boolean connected = true;
                    while (connected){
                        String command = in.readLine();
                        System.out.println("Command received: " + command);
                        if (command == null) {
                            connected = false;
                        } else if (command.equals("shutdown")) {
                            running.set(false);
                            try (Socket tmpSocket = new Socket("localhost", 10101)) {}
                        }
                        // process other commands
                    }
                }
                System.out.println("Client disconnected.");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

更新：更改示例以建立与服务器的连接以使其解除阻塞。

更新： 处理客户端断开连接的示例代码。感谢@user207421 强调了这一点（谢谢）。

更新：更改了示例代码以处理多个客户端套接字/线程。请注意，当您关闭当前刚刚打印到标准错误的套接字时，它将通过一个异常。您可能希望以不同的方式处理它。

更新：您可能还会发现模拟多个客户端连接的代码很有帮助：

public class Clients {
    public static void main(String[] args) {
        Thread thread1 = new Thread(() -> new NormalClient().run());
        Thread thread2 = new Thread(() -> new NormalClient().run());
        Thread thread3 = new Thread(() -> new NormalClient().run());
        Thread thread4 = new Thread(() -> new NormalClient().run());
        Thread thread5 = new Thread(() -> new ShutdownClient().run());
        thread1.start();
        thread2.start();
        thread3.start();
        thread4.start();
        thread5.start();
    }
}

class NormalClient {
    void run() {
        try {
            try (Socket socket = new Socket("localhost", 10101);
                 BufferedWriter out = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()))) {

                for (int i = 0; i < 10; i++) {
                    out.write("hello " + i);
                    out.newLine();
                    out.flush();
                    sleep(1000);
                }
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class ShutdownClient {
    void run() {
        try {
            try (Socket socket = new Socket("localhost", 10101);
                 BufferedWriter out = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()))) {

                sleep(8000);
                out.write("shutdown");
                out.newLine();
                out.flush();
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}