public class ChatServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(9000);
        while (true) {
            try {
                System.out.println("聊天服務(wù)已啟動(dòng)，等待客戶連接....");
                Socket socket = serverSocket.accept();
                System.out.printf("建立了與%s的連接!\n",socket.getRemoteSocketAddress());
                loopReadRequest(socket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static String loopReadRequest(Socket socket) throws IOException {
        InputStreamReader reader = new InputStreamReader(socket.getInputStream());
        StringBuilder sb = new StringBuilder();
        char[] cbuf = new char[256];
        
        // 循環(huán)讀取socket的輸入數(shù)據(jù)流
        while (true) {
            // read方法，讀出內(nèi)容寫入 char 數(shù)組,read 方法會(huì)一直阻塞
            // 直到有輸入內(nèi)容 或 發(fā)生I/O錯(cuò)誤 或 輸入流結(jié)束(對(duì)方關(guān)閉了socket)
            // 正常讀取時(shí)方法會(huì)返回讀取的字符數(shù),當(dāng)輸入流結(jié)束時(shí)(對(duì)方關(guān)閉了socket)方法返回 -1
            int readed = reader.read(cbuf);
   SocketAddress remoteSocketAddress = socket.getRemoteSocketAddress();
   // 客戶端執(zhí)行了socket.close()
            if (readed == -1) {
                System.out.println(remoteSocketAddress + " 斷開(kāi)了連接!");
                reader.close();
                socket.close();
                break;
            }

            String readedStr = new String(cbuf, 0, readed);
            sb.append(readedStr);

      // ready()用來(lái)判斷流是否可被讀取,如果reader緩沖區(qū)不是空則返回true，否則返回false
            if (!reader.ready()) {//reader緩沖區(qū)為空，表示數(shù)據(jù)流已讀完
                // 數(shù)據(jù)流已讀完,此時(shí)向客戶端發(fā)送響應(yīng)
                socket.getOutputStream().write((remoteSocketAddress+"你好，"+sb+"已收到").getBytes());
                System.out.println("收到內(nèi)容:"+sb);
                // 清除sb的內(nèi)容，準(zhǔn)備接收下一個(gè)請(qǐng)求內(nèi)容
                sb.setLength(0);
                System.out.println("等待客戶端消息....");
            }
        }
        return sb.toString();
    }
}

public class ChatClient {
    public static void main(String[] args) {
        try {
            Socket socket = new Socket("localhost", 9000);
            Scanner scanner = new Scanner(System.in);
            while (true) {
                System.out.print(">");
                String line = scanner.nextLine();
                if("".equals(line)){
                    continue;
                }
                if ("quit".equals(line)) {
                    scanner.close();
                    socket.close();
                    break;
                }
                socket.getOutputStream().write(line.getBytes());
                System.out.println(readRequest(socket));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static String readRequest(Socket socket) throws IOException {
        InputStreamReader reader = new InputStreamReader(socket.getInputStream());
        StringBuilder sb = new StringBuilder();
        char[] cbuf = new char[256];
        while (true) {
            int readed = reader.read(cbuf);
            // 讀出內(nèi)容寫入 char 數(shù)組,read 方法會(huì)一直阻塞
            // 直到有輸入內(nèi)容 或 發(fā)生I/O錯(cuò)誤 或 輸入流結(jié)束(對(duì)方關(guān)閉了socket)
            // 正常讀取，方法會(huì)返回讀取的字符數(shù)，而當(dāng)輸入流結(jié)束(對(duì)方關(guān)閉了socket)則返回 -1
            if (readed == -1) {
                System.out.println(socket.getRemoteSocketAddress() + " 斷開(kāi)了連接!");
                reader.close();
                socket.close();
                break;
            }

            String readedStr = new String(cbuf, 0, readed);
            sb.append(readedStr);
            if(!reader.ready()){
                break;
            }
        }
        return sb.toString();
    }
}

public class ChatServer {
    private static Map<String, Socket> connnectedSockets = new ConcurrentHashMap<>();

    public static void main(String[] args) throws IOException {

        // 1、服務(wù)端初始化工作
        ServerSocket serverSocket = new ServerSocket(9000);
        ExecutorService executorService = getExecutorService();

        // 2、主線程- 循環(huán)阻塞接收新的連接請(qǐng)求
        while (true) {
            Socket socket = serverSocket.accept();
            cacheSocket(socket);

            // 3、一個(gè)socket對(duì)應(yīng)一個(gè)讀取任務(wù),交給線程池中的線程執(zhí)行
            // 如果使用fixed線程池,會(huì)操作讀取任務(wù)分配不到線程的情況
            // 現(xiàn)象就是發(fā)送的消息別人收不到（暫存在Socket緩存中）
            executorService.submit(createLoopReadTask(socket));
        }
    }

    private static Runnable createLoopReadTask(Socket socket) {
        return new Runnable() {
            public void run() {
                try {
                    loopReadRequestAndRedirect(socket);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        };
    }

    private static ExecutorService getExecutorService() {
        ExecutorService executorService = Executors.newCachedThreadPool();
        int nThreads = Runtime.getRuntime().availableProcessors();
        nThreads = 1;
        // 如果只設(shè)置一個(gè)線程，那么最先連接進(jìn)來(lái)的客戶端可以發(fā)送消息
        // 因?yàn)槌绦蜃枞x取第一個(gè)socket連接的數(shù)據(jù)流，沒(méi)有其他線程資源去讀后面建立的socket了
        executorService = Executors.newFixedThreadPool(nThreads);
        return executorService;
    }

    private static void cacheSocket(Socket socket) {
        SocketAddress remoteSocketAddress = socket.getRemoteSocketAddress();
        String[] split = remoteSocketAddress.toString().split(":");
        connnectedSockets.put(split[1], socket);
    }

    public static String loopReadRequestAndRedirect(Socket socket) throws IOException {
        InputStreamReader reader = new InputStreamReader(socket.getInputStream());
        StringBuilder sb = new StringBuilder();
        char[] cbuf = new char[256];
        while (true) {
            SocketAddress remoteSocketAddress = socket.getRemoteSocketAddress();
            System.out.println(Thread.currentThread() + "執(zhí)行 " + remoteSocketAddress + "發(fā)送的消息");
            // 讀出內(nèi)容寫入 char 數(shù)組,read 方法會(huì)一直阻塞
            // 直到有輸入內(nèi)容 或 發(fā)生I/O錯(cuò)誤 或 輸入流結(jié)束(對(duì)方關(guān)閉了socket)
            // 正常讀取時(shí)方法會(huì)返回讀取的字符數(shù)，當(dāng)輸入流結(jié)束(對(duì)方關(guān)閉了socket)時(shí)返回 -1
            int readed = reader.read(cbuf);

            if (readed == -1) {
                System.out.println(remoteSocketAddress + " 斷開(kāi)了連接!");
                reader.close();
                socket.close();
                break;
            }

            String readedStr = new String(cbuf, 0, readed);
            sb.append(readedStr);

            //ready()用來(lái)判斷流是否可被讀取,如果reader緩沖區(qū)不是空則返回true，否則返回false
            boolean oneReqeustStreamReaded = !reader.ready();
            if (oneReqeustStreamReaded) {
                String requestContent = sb.toString().trim();
                String prifix = requestContent.substring(0, 2);
                // 單聊
                if ("-c".equals(prifix)) {
                    requestContent = requestContent.substring(3);
                    String port = requestContent.substring(0, requestContent.indexOf(" "));
                    requestContent = requestContent.replaceFirst(port, "");
                    sendToOneSocket(connnectedSockets.get(port), requestContent);
                    // 群聊
                } else {
                    // 向客戶端發(fā)送響應(yīng)
                    socket.getOutputStream().write(("您發(fā)送的消息-'" + sb + "' 已收到").getBytes());
                    sendToAllSocket(sb.toString(), socket);
                }
                sb.setLength(0);
            }
        }
        return sb.toString();
    }

    /**
     * 發(fā)送消息給某個(gè)socket
     *
     * @param socket
     * @param msg
     */
    private static void sendToOneSocket(Socket socket, String msg) {
        // 對(duì)于同一個(gè)socket,同一時(shí)刻只有一個(gè)線程使用它發(fā)送消息
        synchronized (socket) {
            try {
                socket.getOutputStream().write(msg.getBytes("UTF-8"));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 發(fā)送消息給所有的socket
     *
     * @param msg
     */
    private static void sendToAllSocket(String msg, Socket selfSocket) {
        for (String key : connnectedSockets.keySet()) {
            Socket socket = connnectedSockets.get(key);
            if (socket.equals(selfSocket)) {
                continue;
            }
            sendToOneSocket(socket, msg);
        }
    }
}


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

    public void start() throws IOException {
        Socket socket = new Socket("localhost", 9000);
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        
        Runnable readTask = new Runnable() {
            public void run() {
                try {
                    loopReadRequest(socket);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        };
        executorService.submit(readTask);

        Runnable sendMsgTask = new Runnable() {
            public void run() {
                try {
                    Scanner scanner = new Scanner(System.in);
                    while (true) {
                        System.out.print(">");
                        String line = scanner.nextLine();
                        if ("".equals(line)) {
                            continue;
                        }
                        if ("quit".equals(line)) {
                            scanner.close();
                            socket.close();
                            break;
                        }
                        socket.getOutputStream().write(line.getBytes());
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        };
        executorService.submit(sendMsgTask);

    }

    public void loopReadRequest(Socket socket) throws IOException {
        InputStreamReader reader = new InputStreamReader(socket.getInputStream());
        StringBuilder sb = new StringBuilder();
        char[] cbuf = new char[256];
        while (true) {
            int readed = reader.read(cbuf);
            // 讀出內(nèi)容寫入 char 數(shù)組,read 方法會(huì)一直阻塞
            // 直到有輸入內(nèi)容 或 發(fā)生I/O錯(cuò)誤 或 輸入流結(jié)束(對(duì)方關(guān)閉了socket)
            // 正常讀取，方法會(huì)返回讀取的字符數(shù)，而當(dāng)輸入流結(jié)束(對(duì)方關(guān)閉了socket)則返回 -1
            if (readed == -1) {
                System.out.println(socket.getRemoteSocketAddress() + " 斷開(kāi)了連接!");
                reader.close();
                socket.close();
                break;
            }

            String readedStr = new String(cbuf, 0, readed);
            sb.append(readedStr);
            if (!reader.ready()) {
                System.out.println(sb);
                sb.setLength(0);
            }
        }
    }
}