当前位置：移动技术网 > IT编程>开发语言>Java > 高级Java工程师必备 ----- 深入分析 Java IO （一）BIO

高级Java工程师必备 ----- 深入分析 Java IO （一）BIO

2019年06月24日 | 移动技术网IT编程 | 我要评论

bio编程

最原始bio

网络编程的基本模型是c/s模型，即两个进程间的通信。

服务端提供ip和监听端口，客户端通过连接操作想服务端监听的地址发起连接请求，通过三次握手连接，如果连接成功建立，双方就可以通过套接字进行通信。

传统的同步阻塞模型开发中，serversocket负责绑定ip地址，启动监听端口；socket负责发起连接操作。连接成功后，双方通过输入和输出流进行同步阻塞式通信。
最原始bio通信模型图：

存在的问题：

同一时间，服务器只能接受来自于客户端a的请求信息；虽然客户端a和客户端b的请求是同时进行的，但客户端b发送的请求信息只能等到服务器接受完a的请求数据后，才能被接受。(acceptor只有在接受完client1的请求后才能接受client2的请求)
由于服务器一次只能处理一个客户端请求，当处理完成并返回后（或者异常时），才能进行第二次请求的处理。很显然，这样的处理方式在高并发的情况下，是不能采用的。

一请求一线程bio

那有没有方法改进呢? ，答案是有的。改进后bio通信模型图：

此种bio通信模型的服务端，通常由一个独立的acceptor线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理没处理完成后，通过输出流返回应答给客户端，线程销毁。即典型的一请求一应答通宵模型。

代码演示

服务端：

package demo.com.test.io.bio;

import java.io.ioexception;
import java.io.inputstream;
import java.io.outputstream;
import java.net.serversocket;
import java.net.socket;

import demo.com.test.io.nio.niosocketserver;

public class biosocketserver {
   //默认的端口号  
   private static int default_port = 8083;  

   public static void main(string[] args) {
       serversocket serversocket = null;
       try {
           system.out.println("监听来自于"+default_port+"的端口信息");
           serversocket = new serversocket(default_port);
           while(true) {
               socket socket = serversocket.accept();
               socketserverthread socketserverthread = new socketserverthread(socket);
               new thread(socketserverthread).start();
           }
       } catch(exception e) {

       } finally {
           if(serversocket != null) {
               try {
                   serversocket.close();
               } catch (ioexception e) {
                   // todo auto-generated catch block
                   e.printstacktrace();
               }
           }
       }

        //这个wait不涉及到具体的实验逻辑，只是为了保证守护线程在启动所有线程后，进入等待状态
       synchronized (niosocketserver.class) {
           try {
               biosocketserver.class.wait();
           } catch (interruptedexception e) {
               // todo auto-generated catch block
               e.printstacktrace();
           }
       }
   }
}  

class socketserverthread implements runnable {
   private socket socket;
   public socketserverthread (socket socket) {
       this.socket = socket;
   }
   @override
   public void run() {
       inputstream in = null;
       outputstream out = null;
       try {
           //下面我们收取信息
           in = socket.getinputstream();
           out = socket.getoutputstream();
           integer sourceport = socket.getport();
           int maxlen = 1024;
           byte[] contextbytes = new byte[maxlen];
           //使用线程，同样无法解决read方法的阻塞问题，
           //也就是说read方法处同样会被阻塞，直到操作系统有数据准备好
           int reallen = in.read(contextbytes, 0, maxlen);
           //读取信息
           string message = new string(contextbytes , 0 , reallen);

           //下面打印信息
           system.out.println("服务器收到来自于端口：" + sourceport + "的信息：" + message);

           //下面开始发送信息
           out.write("回发响应信息！".getbytes());
       } catch(exception e) {
           system.out.println(e.getmessage());
       } finally {
           //试图关闭
           try {
               if(in != null) {
                   in.close();
               }
               if(out != null) {
                   out.close();
               }
               if(this.socket != null) {
                   this.socket.close();
               }
           } catch (ioexception e) {
               system.out.println(e.getmessage());
           }
       }
   }
}

客户端：

package demo.com.test.io.bio;

import java.io.ioexception;
import java.io.inputstream;
import java.io.outputstream;
import java.net.socket;
import java.net.urldecoder;
import java.util.concurrent.countdownlatch;

public class biosocketclient{
   public static void main(string[] args) throws exception {
       integer clientnumber = 20;
       countdownlatch countdownlatch = new countdownlatch(clientnumber);

       // 分别开始启动这20个客户端,并发访问
       for (int index = 0; index < clientnumber; index++, countdownlatch.countdown()) {
           clientrequestthread client = new clientrequestthread(countdownlatch, index);
           new thread(client).start();
       }

       // 这个wait不涉及到具体的实验逻辑，只是为了保证守护线程在启动所有线程后，进入等待状态
       synchronized (biosocketclient.class) {
           biosocketclient.class.wait();
       }
   }
}



/**
* 一个clientrequestthread线程模拟一个客户端请求。
* @author keep_trying
*/
class clientrequestthread implements runnable {


   private countdownlatch countdownlatch;

   /**
    * 这个线程的编号
    * @param countdownlatch
    */
   private integer clientindex;

   /**
    * countdownlatch是java提供的同步计数器。
    * 当计数器数值减为0时，所有受其影响而等待的线程将会被激活。这样保证模拟并发请求的真实性
    * @param countdownlatch
    */
   public clientrequestthread(countdownlatch countdownlatch , integer clientindex) {
       this.countdownlatch = countdownlatch;
       this.clientindex = clientindex;
   }

   @override
   public void run() {
       socket socket = null;
       outputstream clientrequest = null;
       inputstream clientresponse = null;

       try {
           socket = new socket("localhost",8083);
           clientrequest = socket.getoutputstream();
           clientresponse = socket.getinputstream();

           //等待，直到socketclientdaemon完成所有线程的启动，然后所有线程一起发送请求
           this.countdownlatch.await();

           //发送请求信息
           clientrequest.write(("这是第" + this.clientindex + " 个客户端的请求。 over").getbytes());
           clientrequest.flush();

           //在这里等待，直到服务器返回信息
          system.out.println("第" + this.clientindex + "个客户端的请求发送完成，等待服务器返回信息");
           int maxlen = 1024;
           byte[] contextbytes = new byte[maxlen];
           int reallen;
           string message = "";
           //程序执行到这里，会一直等待服务器返回信息（注意，前提是in和out都不能close，如果close了就收不到服务器的反馈了）
           while((reallen = clientresponse.read(contextbytes, 0, maxlen)) != -1) {
               message += new string(contextbytes , 0 , reallen);
           }
           //string messageencode = new string(message , "utf-8");
           message = urldecoder.decode(message, "utf-8");
           system.out.println("第" + this.clientindex + "个客户端接收到来自服务器的信息:" + message);
       } catch (exception e) {

       } finally {
           try {
               if(clientrequest != null) {
                   clientrequest.close();
               }
               if(clientresponse != null) {
                   clientresponse.close();
               }
           } catch (ioexception e) {

           }
       }
   }
}

存在的问题：

虽然在服务器端，请求的处理交给了一个独立线程进行，但是操作系统通知accept()的方式还是单个的。也就是，实际上是服务器接收到数据报文后的“业务处理过程”可以多线程，但是数据报文的接受还是需要一个一个的来(acceptor只有在接受完client1的请求后才能接受client2的请求)，下文会验证。
在linux系统中，可以创建的线程是有限的。我们可以通过cat /proc/sys/kernel/threads-max命令查看可以创建的最大线程数。当然这个值是可以更改的，但是线程越多，cpu切换所需的时间也就越长，用来处理真正业务的需求也就越少。
另外，如果您的应用程序大量使用长连接的话，线程是不会关闭的。这样系统资源的消耗更容易失控。

伪异步i/o编程

为了改进这种一连接一线程的模型，我们可以使用线程池来管理这些线程，实现1个或多个线程处理n个客户端的模型（但是底层还是使用的同步阻塞i/o），通常被称为“伪异步i/o模型“。

伪异步i/o模型图：

代码演示

只给出服务端，客户端和上面相同

package demo.com.test.io.bio;

import java.io.ioexception;
import java.io.inputstream;
import java.io.outputstream;
import java.net.serversocket;
import java.net.socket;
import java.util.concurrent.executorservice;
import java.util.concurrent.executors;

import demo.com.test.io.nio.niosocketserver;

public class biosocketserverthreadpool {
   //默认的端口号  
   private static int default_port = 8083;  
   //线程池 懒汉式的单例  
   private static executorservice executorservice = executors.newfixedthreadpool(60);  

   public static void main(string[] args) {
       serversocket serversocket = null;
       try {
           system.out.println("监听来自于"+default_port+"的端口信息");
           serversocket = new serversocket(default_port);
           while(true) {
               socket socket = serversocket.accept();
               //当然业务处理过程可以交给一个线程（这里可以使用线程池）,并且线程的创建是很耗资源的。
               //最终改变不了.accept()只能一个一个接受socket的情况,并且被阻塞的情况
               socketserverthreadpool socketserverthreadpool = new socketserverthreadpool(socket);
               executorservice.execute(socketserverthreadpool);
           }
       } catch(exception e) {

       } finally {
           if(serversocket != null) {
               try {
                   serversocket.close();
               } catch (ioexception e) {
                   // todo auto-generated catch block
                   e.printstacktrace();
               }
           }
       }

        //这个wait不涉及到具体的实验逻辑，只是为了保证守护线程在启动所有线程后，进入等待状态
       synchronized (niosocketserver.class) {
           try {
               biosocketserverthreadpool.class.wait();
           } catch (interruptedexception e) {
               // todo auto-generated catch block
               e.printstacktrace();
           }
       }
   }
}  

class socketserverthreadpool implements runnable {
   private socket socket;
   public socketserverthreadpool (socket socket) {
       this.socket = socket;
   }
   @override
   public void run() {
       inputstream in = null;
       outputstream out = null;
       try {
           //下面我们收取信息
           in = socket.getinputstream();
           out = socket.getoutputstream();
           integer sourceport = socket.getport();
           int maxlen = 1024;
           byte[] contextbytes = new byte[maxlen];
           //使用线程，同样无法解决read方法的阻塞问题，
           //也就是说read方法处同样会被阻塞，直到操作系统有数据准备好
           int reallen = in.read(contextbytes, 0, maxlen);
           //读取信息
           string message = new string(contextbytes , 0 , reallen);

           //下面打印信息
           system.out.println("服务器收到来自于端口：" + sourceport + "的信息：" + message);

           //下面开始发送信息
           out.write("回发响应信息！".getbytes());
       } catch(exception e) {
           system.out.println(e.getmessage());
       } finally {
           //试图关闭
           try {
               if(in != null) {
                   in.close();
               }
               if(out != null) {
                   out.close();
               }
               if(this.socket != null) {
                   this.socket.close();
               }
           } catch (ioexception e) {
               system.out.println(e.getmessage());
           }
       }
   }
}

服务器端的执行效果

在 socket socket = serversocket.accept(); 处打了断点，有20个客户端同时发出请求，可服务端还是一个一个的处理，其它线程都处于阻塞状态

阻塞的问题根源

那么重点的问题并不是“是否使用了多线程、或是线程池”，而是为什么accept()、read()方法会被阻塞。api文档中对于 serversocket.accept() 方法的使用描述：

listens for a connection to be made to this socket and accepts it. the method blocks until a connection is made.

服务器线程发起一个accept动作，询问操作系统是否有新的socket套接字信息从端口xx发送过来。

注意，是询问操作系统。也就是说socket套接字的io模式支持是基于操作系统的，那么自然同步io/异步io的支持就是需要操作系统级别的了。如下图：

如果操作系统没有发现有套接字从指定的端口xx来，那么操作系统就会等待。这样serversocket.accept()方法就会一直等待。这就是为什么accept()方法为什么会阻塞：它内部的实现是使用的操作系统级别的同步io。

阻塞io 和非阻塞io
这两个概念是程序级别的。主要描述的是程序请求操作系统io操作后，如果io资源没有准备好，那么程序该如何处理的问题：前者等待；后者继续执行（并且使用线程一直轮询，直到有io资源准备好了）
同步io 和非同步io
这两个概念是操作系统级别的。主要描述的是操作系统在收到程序请求io操作后，如果io资源没有准备好，该如何处理相应程序的问题：前者不响应，直到io资源准备好以后；后者返回一个标记（好让程序和自己知道以后的数据往哪里通知），当io资源准备好以后，再用事件机制返回给程序。

您可能感兴趣的文章:

如对本文有疑问，点击进行留言回复！！

浅谈Java分布式架构下如何实现分布式锁

01分布式锁运用场景　　互联网秒杀，抢优惠卷，接口幂等性校验。咱们以互联网秒杀为例。　@restcontroller　　@slf4j　　publicclassi... [阅读全文]
详解SpringBoot修改启动端口server.port的四种方式

方式一: 配置文件 application.propertiesserver.port=7788方式二: java启动命令# 以应用参数的方式java -jar... [阅读全文]
Spring AOP里的静态代理和动态代理用法详解

什么是代理？　　为某一个对象创建一个代理对象，程序不直接用原本的对象，而是由创建的代理对象来控制原对象，通过代理类这中间一层，能有效控制对委托类对象的直接访问，... [阅读全文]
java实现通讯录管理系统

本文实例为大家分享了java实现通讯录管理系统的具体代码，供大家参考，具体内容如下完成项目的流程：1.根据需求，确定大体方向 2.功能模块分析 3.界面实现 4... [阅读全文]
Java中数据库常用的两把锁之乐观锁和悲观锁

在写入数据库的时候需要有锁，比如同时写入数据库的时候会出现丢数据，那么就需要锁机制。　　数据锁分为乐观锁和悲观锁，那么它们使用的场景如下：　　1. 乐观锁适用于... [阅读全文]
git在idea中的冲突解决方法(非常重要)

1、git冲突的场景情景一：多个分支代码合并到一个分支时；情景二：多个分支向同一个远端分支推送代码时；实际上，push操作即是将本地代码merge到远端... [阅读全文]
Java代码生成器的制作流程详解

1. 前言前几天写了篇关于mybatis plus代码生成器的文章，不少同学私下问我这个代码生成器是如何运作的，为什么要用到一些模板引擎，所以今天来说明下代码生... [阅读全文]
Java中生产者消费者问题总结

生产者-消费者算是并发编程中常见的问题。依靠缓冲区我们可以实现生产者与消费者之间的解耦。生产者只管往缓冲区里面放东西，消费者只管往缓冲区里面拿东西。这样我们避免... [阅读全文]
Java 编程中十个处理异常的建议

一、尽量不要使用e.printstacktrace(),而是使用log打印。反例:try{ // do what you want }catch(excepti... [阅读全文]
Spring boot热部署devtools过程解析

1 pom.xml文件注：热部署功能spring-boot-1.3开始有的<dependency> <gr... [阅读全文]