tcp 服务器端 的数据发送与接收并行

tcp 服务器端;

即 在 TCPServer  中 有一个 客户端监听类ClientListener   ，

使用 do-while()  不停的监听 客户端的连接，即 监听到一个客户端的到来，就创建一个

ClientHandler客户端处理类（在这个类中 有两个 线程类:ClientReadHandler  和ClientWriteHandler ）

这样就实现了 数据发送与接收 并行。

 

 

 

 

1.首先在 TCPServer 的start 函数中  先 启动  ClientListener【客户端监听类】：

 

2. ClientListener  类：

在其run() 函数中 使用 do while()  不停的监听客户端，

client = server.accept(); 是阻塞 函数

直到 有客户端连接才继续执行，

当有 客户端连接 时，就创建异步线程（将客户端消息的读取和发送由别的线程完成）：

##################################################################

private class ClientListener extends Thread {
    private ServerSocket server;
    private boolean done = false;

    private ClientListener(int port) throws IOException {
        server = new ServerSocket(port);
        System.out.println("服务器信息：" + server.getInetAddress() + " P:" + server.getLocalPort());
    }
    @Override
    public void run() {
        super.run();

        System.out.println("服务器准备就绪～");
        // 等待客户端连接
        do {
            // 得到客户端
            Socket client;
            try {
                client = server.accept();
            } catch (IOException e) {
                continue;
            }
            try {
                // 客户端构建异步线程
                ClientHandler clientHandler = new ClientHandler(client,
                        handler -> clientHandlerList.remove(handler));
                // 读取数据并打印
                clientHandler.readToPrint();
                clientHandlerList.add(clientHandler);
            } catch (IOException e) {
                e.printStackTrace();
                System.out.println("客户端连接异常：" + e.getMessage());
            }
        } while (!done);

        System.out.println("服务器已关闭！");
    }

    void exit() {
        done = true;
        try {
            server.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

##################################################################

 

 

3. ClientHandler 类：

在这个类中，将消息的读取 和 发送 分成了两个线程：

private final ClientReadHandler readHandler;

private final ClientWriteHandler writeHandler;

 

##################################################################

package server.handle;

import clink.net.qiujuer.clink.utils.CloseUtils;

import java.io.*;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ClientHandler {
    private final Socket socket;
    private final ClientReadHandler readHandler;
    private final ClientWriteHandler writeHandler;
    private final CloseNotify closeNotify;

    public ClientHandler(Socket socket, CloseNotify closeNotify) throws IOException {
        this.socket = socket;
        this.readHandler = new ClientReadHandler(socket.getInputStream());
        this.writeHandler = new ClientWriteHandler(socket.getOutputStream());
        this.closeNotify = closeNotify;
        System.out.println("新客户端连接：" + socket.getInetAddress() +
                " P:" + socket.getPort());
    }

    public void exit() {
        readHandler.exit();
        writeHandler.exit();
        CloseUtils.close(socket);
        System.out.println("客户端已退出：" + socket.getInetAddress() +
                " P:" + socket.getPort());
    }

    public void send(String str) {
        writeHandler.send(str);
    }

    public void readToPrint() {
        readHandler.start();
    }

    private void exitBySelf() {
        exit();
        closeNotify.onSelfClosed(this);
    }

    public interface CloseNotify {
        void onSelfClosed(ClientHandler handler);
    }

    class ClientReadHandler extends Thread {
        private boolean done = false;
        private final InputStream inputStream;

        ClientReadHandler(InputStream inputStream) {
            this.inputStream = inputStream;
        }

        @Override
        public void run() {
            super.run();
            try {
                // 得到输入流，用于接收数据
                BufferedReader socketInput = new BufferedReader(new InputStreamReader(inputStream));

                do {
                    // 客户端拿到一条数据
                    String str = socketInput.readLine();
                    if (str == null) {
                        System.out.println("客户端已无法读取数据！");
                        // 退出当前客户端
                        ClientHandler.this.exitBySelf();
                        break;
                    }
                    // 打印到屏幕
                    System.out.println(str);
                } while (!done);
            } catch (Exception e) {
                if (!done) {
                    System.out.println("连接异常断开");
                    ClientHandler.this.exitBySelf();
                }
            } finally {
                // 连接关闭
                CloseUtils.close(inputStream);
            }
        }

        void exit() {
            done = true;
            CloseUtils.close(inputStream);
        }
    }

    class ClientWriteHandler {
        private boolean done = false;
        private final PrintStream printStream;
        private final ExecutorService executorService;

        ClientWriteHandler(OutputStream outputStream) {
            this.printStream = new PrintStream(outputStream);
            this.executorService = Executors.newSingleThreadExecutor();
        }

        void exit() {
            done = true;
            CloseUtils.close(printStream);
            executorService.shutdownNow();
        }

        void send(String str) {
            executorService.execute(new WriteRunnable(str));
        }

        class WriteRunnable implements Runnable {
            private final String msg;

            WriteRunnable(String msg) {
                this.msg = msg;
            }

            @Override
            public void run() {
                if (ClientWriteHandler.this.done) {
                    return;
                }

                try {
                    ClientWriteHandler.this.printStream.println(msg);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

 

##################################################################