一、BIO编程模型概述

Acceptor：接受客户的请求
Client：客户端发送的请求

问题引出：

如果像前一个Socket和ServerSocket的模型一样设计
在Client发送请求的时候，Acceptor接受了以后，就不能再接受更多的请求了，就变成了自言自语模式了

解决方法：

因此引出了的Handler来处理请求，于是再发送请求，Acceptor就可以接受请求了。
形成模型如下图：
image.png
这样的模型即BIO传统模型

主要功能

  1. 基于BIO模型
  2. 支持多人同时在线
  3. 每个用户的发言都被转发给其他用户

    架构设计

    首先需要一个主线程来做Acceptor
    其次需要另外一个线程来做Handler，来进行数据处理和数据据读写
    每一个客户都要对应一个Handler
    服务器端需要存储所有客户信息，才可以把消息广播给所有客户
    客户端不能在输入的时候阻塞接收信息

    二、代码实现

    客户端：

  1. package demo2.client;
  3. import java.io.*;
  4. import java.net.Socket;
  6. public class ChatClient {
  7.     private final String DEFAULT_IP="localhost";
  8.     private final int DEFAULT_PORT=9999;
  9.     private final String QUIT="quit";
  10.     private BufferedReader reader;
  11.     private BufferedWriter writer;
  12.     private Socket socket;
  13.     public void send(String msg) throws IOException {
  14.         if(msg!=null&&!socket.isOutputShutdown()){
  15.             writer.write(msg+"\n");
  16.             writer.flush();
  17.         }
  18.     }
  19.     public String receive() throws IOException {
  20.         String msg=null;
  21.         if (!socket.isInputShutdown()){
  22.             msg=reader.readLine();
  23.         }
  24.         return msg;
  25.     }
  27.     public boolean isQuit(String msg){
  28.         return QUIT.equalsIgnoreCase(msg);
  29.     }
  30.     public void close(){
  31.         try {
  32.             if (writer != null) {
  33.                 writer.close();
  34.             }
  35.             if (reader != null) {
  36.                 reader.close();
  37.             }
  38.             if (socket != null) {
  39.                 System.out.println("客户端["+socket.getPort()+"]关闭");
  40.                 socket.close();
  41.             }
  42.         } catch (IOException e) {
  43.             e.printStackTrace();
  44.         }
  45.     }
  46.     public void start(){
  47.         try {
  48.             socket=new Socket(DEFAULT_IP,DEFAULT_PORT);
  49.             System.out.println("客户端["+DEFAULT_IP+"]成功启动");
  50.             writer=new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
  51.             reader=new BufferedReader(new InputStreamReader(socket.getInputStream()));
  52.             new Thread(new UserInputHander(this)).start();
  53.             String msg=null;
  54.             while ((msg=receive())!=null){
  55.                 System.out.println(msg);
  56.             }
  57.         } catch (IOException e) {
  58.             e.printStackTrace();
  59.         }finally {
  60.             close();
  61.         }
  62.     }
  63. }

线程部分（与输出不同线程）：

  1. package demo2.client;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  7. public class UserInputHander implements Runnable {
  8.     private ChatClient chatClient;
  9.     private BufferedReader reader;
  11.     public UserInputHander(ChatClient chatClient) {
  12.         this.chatClient = chatClient;
  13.     }
  15.     @Override
  16.     public void run() {
  17.         reader = new BufferedReader(new InputStreamReader(System.in));
  19.         try {
  20.             while (true) {
  21.                 String msg = reader.readLine();
  22.                 chatClient.send(msg);
  23.                 if (chatClient.isQuit(msg)) {
  24.                     break;
  25.                 }
  26.             }
  27.         } catch (IOException e) {
  28.             e.printStackTrace();
  29.         } finally {
  30.             if (reader != null) {
  31.                 try {
  32.                     reader.close();
  33.                 } catch (IOException e) {
  34.                     e.printStackTrace();
  35.                 }
  36.             }
  37.         }
  38.     }
  39. }

主函数：

  1. package demo2.client;
  2. public class ClientMain {
  3.     public static void main(String[] args) {
  4.         ChatClient chatClient=new ChatClient();
  5.         chatClient.start();
  6.     }
  7. }

服务器：

  1. package demo2.server;
  3. import java.io.*;
  4. import java.net.ServerSocket;
  5. import java.net.Socket;
  6. import java.util.HashMap;
  7. import java.util.Map;
  9. public class ChatServer {
  10.     private ServerSocket serverSocket;
  11.     private final int DEFAULT_PORT=9999;
  12.     private final String QUIT="quit";
  13.     private Map<Integer, Writer> connectedClients;
  14.     public ChatServer(){
  15.         connectedClients=new HashMap<>();
  16.     }
  17.     public synchronized void addClient(Socket socket) throws IOException {
  18.         if(socket!=null){
  19.             int port=socket.getPort();
  20.             BufferedWriter writer=new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
  21.             connectedClients.put(port,writer);
  22.             System.out.println("客户端["+port+"]已连接到服务器");
  23.         }
  24.     }
  25.     public synchronized void removeClient(Socket socket) throws IOException {
  26.         if(socket!=null) {
  27.             int port=socket.getPort();
  28.             if (connectedClients.containsKey(port)) {
  29.                 connectedClients.get(port).close();
  30.                 connectedClients.remove(port);
  31.                 System.out.println("客户端[" + port + "]已断开连接");
  32.             }
  33.         }
  34.     }
  35.     public synchronized void forwardMessage(Socket socket,String msg) throws IOException {
  36.         if (socket!=null) {
  37.             for (Integer port : connectedClients.keySet()) {
  38.                 if (port != socket.getPort()) {
  39.                     Writer writer = connectedClients.get(port);
  40.                     writer.write(msg);
  41.                     writer.flush();
  42.                 }
  43.             }
  44.         }
  45.     }
  46.     public synchronized void shutdownClient() throws IOException {
  47.         if (serverSocket!=null){
  48.             serverSocket.close();
  49.             System.out.println("服务器已关闭");
  50.         }
  51.     }
  52.     public boolean isQuit(String msg){
  53.         return QUIT.equalsIgnoreCase(msg);
  54.     }
  55.     public void start(){
  56.         try {
  57.             serverSocket=new ServerSocket(DEFAULT_PORT);
  58.             System.out.println("启动服务器["+DEFAULT_PORT+"]成功");
  59.             while (true){
  60.                 Socket socket=serverSocket.accept();
  61.                 new Thread(new ChatHander(socket,this)).start();
  62.             }
  63.         } catch (IOException e) {
  64.             e.printStackTrace();
  65.         }finally {
  66.             try {
  67.                 shutdownClient();
  68.             } catch (IOException e) {
  69.                 e.printStackTrace();
  70.             }
  71.         }
  72.     }
  73. }

线程部份（处理用户信息）：

  1. package demo2.server;
  4. import java.io.BufferedReader;
  5. import java.io.IOException;
  6. import java.io.InputStreamReader;
  7. import java.net.Socket;
  9. public class ChatHander implements Runnable{
  10.     private ChatServer chatServer;
  11.     private Socket socket;
  12.     private BufferedReader reader;
  13.     public ChatHander(Socket socket,ChatServer chatServer){
  14.         this.chatServer=chatServer;
  15.         this.socket=socket;
  16.     }
  18.     @Override
  19.     public void run() {
  20.         try {
  21.             chatServer.addClient(socket);
  22.             reader=new BufferedReader(new InputStreamReader(socket.getInputStream()));
  23.             String msg=null;
  24.             int port=socket.getPort();
  25.             while ((msg=reader.readLine())!=null) {
  26.                 if (chatServer.isQuit(msg)) {
  27.                     break;
  28.                 }
  29.                 String nwmsg = "客户端[" + port + "]：" + msg + "\n";
  30.                 System.out.print(nwmsg);
  31.                 chatServer.forwardMessage(socket, nwmsg);
  32.             }
  33.         } catch (IOException e) {
  34.             e.printStackTrace();
  35.         }finally {
  36.             try {
  37.                 if (socket != null) {
  38.                     chatServer.removeClient(socket);
  39.                 }
  40.                 if (reader != null) {
  41.                     reader.close();
  42.                 }
  43.             } catch (IOException e) {
  44.                 e.printStackTrace();
  45.             }
  46.         }
  47.     }
  48. }

主函数：

  1. package demo2.server;
  2. public class ServerMain {
  3.     public static void main(String[] args) {
  4.         ChatServer chatServer=new ChatServer();
  5.         chatServer.start();
  6.     }
  7. }

三、代码的效果

image.png

四、伪异步IO编程

随着客户端增加和退出，会产生线程的调度和资源的浪费，回忆线程池的知识，这里引入线程池的做法来解决资源的问题。

功能建模：

image.png
如果Client4用户想要加入群聊时，线程池却没有空余线程，Client则等待其他用户退出时才可以加入。

代码实现：

这里我只对服务器端进行修改，因为客户端的资源浪费少之又少，就不做阐述了。
在ChatServer中增加属性：
private ExecutorService executorService;
修改构造方法：
设置5个线程。

  1. public ChatServer(){
  2.         executorService= Executors.newFixedThreadPool(5);
  3.         connectedClients=new HashMap<>();
  4.     }

修改start方法：

  1. public void start(){
  2.         try {
  3.             serverSocket=new ServerSocket(DEFAULT_PORT);
  4.             System.out.println("启动服务器["+DEFAULT_PORT+"]成功");
  5.             while (true){
  6.                 Socket socket=serverSocket.accept();
  7.                 executorService.execute(new ChatHander(socket,this));
  8.             }
  9.         } catch (IOException e) {
  10.             e.printStackTrace();
  11.         }finally {
  12.             try {
  13.                 shutdownClient();
  14.             } catch (IOException e) {
  15.                 e.printStackTrace();
  16.             }
  17.         }
  18.     }

这样我们就实现了简单的BIO的多人聊天室。