BIO剖析_Java

BIO

BIO模式下的多发与多收消息
BIO模式下接收多个客户端
- 概述
- 代码实现
伪异步I/O编程
- 概述
- 代码实现
BIO实现任意类型文件上传
基于BIO模式下的端口转发思想
BIO即时通讯

BIO模式下的多发与多收消息

Server

public class Server {
    public static void main(String[] args) {
        try {
            System.out.println("服务端启动 ---- ");
            // 1. 定一个ServerSocket对象进行服务端的端口注册
            ServerSocket serverSocket = new ServerSocket(9999);
            // 2. 监听客户端的Socket链接请求
            Socket socket = serverSocket.accept();
            // 3. 从Socket获取字节输入流对象
            InputStream inputStream = socket.getInputStream();
            // 4. 把字节输入流包转成一个缓冲字符输入流
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
            String msg;
            while ((msg = bufferedReader.readLine()) != null) { // while 读取完一行之后  此时Client已经断开 抛异常Connect reset
                System.out.println(msg);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

client

public class Client {
    public static void main(String[] args) {
        try {
            // 1. 创建Socket对象请求服务端的连接
            Socket socket = new Socket("127.0.0.1", 9999);
            // 2. 从Socket获取一个字节输出流
            OutputStream outputStream = socket.getOutputStream();
            // 3. 把字节输出流包转成一个打印流
            PrintStream printStream = new PrintStream(outputStream);
            Scanner scanner = new Scanner(System.in);
            while (true){
                System.out.println("请说：");
                String s = scanner.nextLine();
                printStream.println(s);
                printStream.flush();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

BIO模式下接收多个客户端

演示：Idea中开启服务端，同时开启两个客户端
注意： idea中开启多个客户端需要配置允许并发访问

现象：

问题：
服务端只接受了一个客户端的请求
分析：
服务端中开启了一个线程接受并处理，无法处理另一个socket
Socket socket = serverSocket.accept();

概述

在上述案例中，一个服务器只能接受一个客户端的通信请求，那么如果服务端需要处理很多个客户端的消息通信请求应该如何处理，此时我们就需要在服务端引入线程了，也就是说客户端每发一个请求，服务端就创建一个新的线程来处理这个客户端的请求，这样就实现了一个客户端一个线程的模型，图解模式如下：

代码实现

server

public class Server {
    public static void main(String[] args) {
        try {
            // 1. 注册端口
            ServerSocket ss = new ServerSocket(9999);
            System.out.println("服务端启动了 --");
            // 2. 定义一个死循环，负责不断的接收客户端的Socket连接请求
            while (true) {
                Socket socket = ss.accept(); // 阻塞
                System.out.println(Thread.currentThread().getId() + ": 主线程 -- ");
                // 3. 创建一个独立的线程处理这个客户端的请求
                new ServerThreadReader(socket).start();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}





public class ServerThreadReader extends Thread{
    private Socket socket;
    public ServerThreadReader(Socket socket){
        this.socket = socket;
    }
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getId() + ":子线程 --");
        try {
            InputStream is = this.socket.getInputStream();
            BufferedReader br = new BufferedReader(new InputStreamReader(is));
            String msg = null;
            while ((msg = br.readLine()) != null){
                System.out.println(msg);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

client

public class Client {
    public static void main(String[] args) {
        try {
            Socket socket = new Socket("127.0.0.1", 9999);
            OutputStream os = socket.getOutputStream();
            PrintWriter pw = new PrintWriter(os);
            Scanner sc = new Scanner(System.in);
            while (true) {
                System.out.println("请说：");
                String s = sc.nextLine();
                pw.println(s);
                pw.flush();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

小结

每个Socket接收到，都会创建一个线程，线程的竞争、切换上下文影响性能
每个线程都会占用栈空间和cpu资源
并不是每个socket都进行io操作，无意义的线程处理
客户端的并发访问增加时，服务端将呈现1:1的线程开销，访问量越大，系统将发生线程栈溢出，线程创建失败，最终导致进程宕机或僵死，从而不能对外提供服务

伪异步I/O编程概述

在上述案例中：客户端的并发访问增加时，服务端将呈现1:1的线程开销，访问量越大，系统将发生线程栈溢出，线程创建失败，最终导致进程宕机或者僵死，从而不能提供服务。
接下来我们采用一个伪异步I/O的通信框架，采用线程池和任务队列实现，当客户端接入时，将客户端的socket封装成一个Task（该任务实现java.lang。Runnable线程任务接口）交给后端的线程池中进行处理。JDK的线程池维护一个消息队列和N个活跃的线程，对消息中的Socket任务进行处理，由线程池可以设置消息队列的大小和最大线程数，因此，他的资源占用是可以控制的，无论多少个客户端并发访问，都不会导致资源的耗尽和宕机

代码实现

server

public class Server {
    public static void main(String[] args) {
        try {
            ServerSocket sc = new ServerSocket(9999);
            // 1. 创建线程池对象
            HandlerSocketServerPool serverPool = new HandlerSocketServerPool(10, 20);
            while (true){
                Socket socket = sc.accept();
                // 2. 封装成任务对象
                SocketRunnable socketRunnable = new SocketRunnable(socket);
                // 3. 交由线程池处理（当一个socket连接时，封装成任务对象，交由线程池处理，提交任务操作，让线程池去触发run方法）
                serverPool.execute(socketRunnable);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

threadPool

public class HandlerSocketServerPool {
    // 1. 创建一个线程池的成员变量用于存储一个线程池对象
    private ExecutorService executorService;

    
    public HandlerSocketServerPool(int maxThreadNum, int queueSize) {
        executorService = new ThreadPoolExecutor(3,
                maxThreadNum, 120,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue(queueSize));
    }

    
    public void execute(Runnable target){
        executorService.execute(target);
    }

}

task

public class SocketRunnable implements Runnable {
    private Socket socket;
    public SocketRunnable(Socket socket){
        this.socket = socket;
    }
    @Override
    public void run() {
        try {
            InputStream is = this.socket.getInputStream();
            BufferedReader br = new BufferedReader(new InputStreamReader(is));
            String msg = null;
            while ((msg = br.readLine()) != null) {
                System.out.println(msg);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

小结

伪异步io采用了线程池实现，因此避免了为每个请求创建一个独立线程造成了线程资源耗尽问题，但由于底层依然是采用的异步阻塞模型，因此无法从根本上解决问题
如果单个消息处理缓慢，或者服务器线程中的全部线程都被阻塞，那么后续socket的io消息将在队列中排队。新的Socket请求将被拒绝，客户端发生大量连接超时

BIO实现任意类型文件上传

client

public class Client {
    public static void main(String[] args) {
        try {
            // 1. 请求与服务端的Socket连接
            Socket socket = new Socket("127.0.0.1", 9999);
            // 2. 把字节输出流封装成一个数据输出流
            DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
            // 3. 发送文件后缀
            dos.writeUTF(".png");
            // 4. 发送文件
            FileInputStream fis = new FileInputStream("/Users/mac/Desktop/111.png");
            byte[] buffer = new byte[1024];
            int len;
            while ((len = fis.read(buffer)) > 0) {
                dos.write(buffer,0,len);
            }
            dos.flush();
            socket.isOutputShutdown(); // 通知服务端这边数据发送完毕

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

server

public class Server {
    public static void main(String[] args) {
        try {
            ServerSocket ss = new ServerSocket(9999);
            while (true) {
                Socket socket = ss.accept();
                new ServerReaderThread(socket).start();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}




public class ServerReaderThread extends Thread{
    private Socket socket;
    public ServerReaderThread(Socket socket){
        this.socket = socket;
    }
    @Override
    public void run() {
        try {
            // 1. 得到一个数据输入流读取客户端发来的数据
            DataInputStream dis = new DataInputStream(this.socket.getInputStream());
            // 2. 获取文件类型
            String suffix = dis.readUTF();
            System.out.println("服务端已成功接收文件类型："+suffix);
            // 3. 定义一个字节输出管道负责把客户端发来的数据写进去
            OutputStream os = new FileOutputStream("/Users/mac/IdeaProjects/spring-demos/netty-demo/src/com/zhj/static/" + UUID.randomUUID() + suffix);
            // 4. 从数据输入流中读取数据，写到字节输出流去
            byte[] buffer = new byte[1024];
            int len;
            while ((len = dis.read(buffer)) > 0) {
                os.write(buffer,0,len);
            }
            os.close();
            System.out.println("文件接收完毕！");

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

基于BIO模式下的端口转发思想

需求：需要实现一个客户端的消息可以发送给所有的客户端去接收

public class Server {
    // 定义一个静态集合
    public static List allSocketOnLine = new ArrayList();

    public static void main(String[] args) {
        try {
            ServerSocket ss = new ServerSocket(9999);
            while (true){

                Socket socket = ss.accept();
                // 把登录的客户端socket存入到一个在线的集合中去
                allSocketOnLine.add(socket);
                // 为当前登录成功的socket分配一个独立的线程来处理
                new ServerReaderThread(socket).start();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}





public class ServerReaderThread extends Thread{
    private Socket socket;
    public ServerReaderThread(Socket socket){
        this.socket = socket;
    }
    @Override
    public void run() {
        try {
            BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String msg;
            while ((msg = br.readLine()) != null) {
                // 发送给所有的在线socket
                sendMsgToAllClient(msg);
            }
        } catch (Exception e) {
            System.out.println("有人下线了");
            Server.allSocketOnLine.remove(socket);
        }

    }

    
    private void sendMsgToAllClient(String msg) throws IOException {
        for (Socket sk : Server.allSocketOnLine) {
            PrintWriter pw = new PrintWriter(sk.getOutputStream());
            pw.println(msg);
            pw.flush();
        }
    }
}

BIO即时通讯

参考： https://blog.csdn.net/unique_perfect/article/details/115108640

BIO剖析

Java相关栏目本月热门文章