阻塞式IO通信

Last updated 4 years ago

Was this helpful?

阻塞式IO通信

一、BIO通信介绍

网络编程的基本模型是Client/Server模型，也就是两个进程之间进行相互通信，其中服务端提供位置信息（绑定的IP地址和监听端口），客户端通过连接操作向服务端监听的地址发起连接请求，通过三次握手建立连接，如果连接建立成功，双方就可以通过网络套接字（Socket）进行通信。在基于传统同步阻塞模型开发中，ServerSocket负责绑定IP地址，启动监听端口；Socket负责发起连接操作。连接成功之后，双方通过输入和输出流进行同步阻塞式通信。

先看一下BIO通信模型，采用BIO通信模型的服务端，通常由一个独立的Acceptor线程负责监听客户端的连接，它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理，处理完成之后，通过输出流返回应答给客户端，线程销毁。这就是典型的一请求一应答通信模型。

该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈1：1的正比关系，由于线程是Java虚拟机非常宝贵的系统资源，当线程数膨胀之后，系统的性能将急剧下降，随着并发访问量的继续增大，系统会发生线程堆栈溢出、创建新线程失败等问题，并最终导致进程宕机或者僵死，不能对外提供服务。

二、代码实例

2.1 服务端

TimeServer.java 主线程，监听端口的线程

package com.wangjun.io.bio;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * 同步阻塞式IO创建时间服务器
 * @author wangjun
 * @date 2020-03-25
 * @version 1.0
 */
public class TimeServer {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        if(args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                //port采用默认值
            }
        }

        ServerSocket server = null;
        try {
            server = new ServerSocket(port);
            System.out.println("server started, port:" + port);
            Socket socket = null;
            while(true) {
                socket = server.accept();
                //启动一个新线程处理请求
                System.out.println("start a thread, hostName:" + socket.getInetAddress().getHostName());
                new Thread(new TimeServerHandler(socket)).start();
            }
        } finally {
            if(server != null) {
                System.out.println("server close");
                server.close();
                server = null;
            }
        }
    }
}

TimeServerHandler.java 子线程，处理客户端请求的线程

package com.wangjun.io.bio;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.Date;

/**
 * @author wangjun
 * @date 2020-03-25
 * @version 1.0
 */
public class TimeServerHandler implements Runnable {

    private Socket socket;
    public TimeServerHandler(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() {
        BufferedReader in = null;
        PrintWriter out = null;
        try {
            in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
            out = new PrintWriter(this.socket.getOutputStream(), true);
            String currentTime =  null;
            String body = null;
            while(true) {
                System.out.println("start read data");
                body = in.readLine();
                if(body == null) {
                    System.out.println("body is null");
                    break;
                }
                System.out.println("server receive order:" + body);
                currentTime = "QUERY TIME ORDER".equals(body)? new Date(System.currentTimeMillis()).toString(): "BAD ORDER";
                out.println(currentTime);
            }
        } catch (Exception e) {
            e.printStackTrace();
            if(in != null) {
                try {
                    in.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
            if(out != null) {
                out.close();
                out = null;
            }
            if(this.socket != null) {
                try {
                    this.socket.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
                this.socket = null;
            }
        }
    }

}

启动TimeServer，可以看到打印：

server started, port:8080

此时因为没有客户端接入，主线程阻塞在socket = server.accept();通过Java VisualVM打印线程堆栈，可以看到主程序确实阻塞在accept操作上：

"main" #1 prio=5 os_prio=0 tid=0x00000000053a3000 nid=0x618 runnable [0x000000000539f000]
   java.lang.Thread.State: RUNNABLE
        at java.net.DualStackPlainSocketImpl.accept0(Native Method)
        at java.net.DualStackPlainSocketImpl.socketAccept(Unknown Source)
        at java.net.AbstractPlainSocketImpl.accept(Unknown Source)
        at java.net.PlainSocketImpl.accept(Unknown Source)
        - locked <0x000000076b6e0dc8> (a java.net.SocksSocketImpl)
        at java.net.ServerSocket.implAccept(Unknown Source)
        at java.net.ServerSocket.accept(Unknown Source)
        at com.wangjun.io.bio.TimeServer.main(TimeServer.java:30)

   Locked ownable synchronizers:
        - None

"VM Thread" os_prio=2 tid=0x000000001f109000 nid=0x39dc runnable

2.2 客户端

TimeClient.java 客户端请求socket通信

package com.wangjun.io.bio;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

/**
 * @author wangjun
 * @date 2020-03-25
 * @version 1.0
 */
public class TimeClient {
    public static void main(String[] args) {
        int port = 8080;
        Socket socket = null;
        BufferedReader in = null;
        PrintWriter out = null;
        try {
            socket = new Socket("127.0.0.1", port);
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream());
            out.println("QUERY TIME ORDER");
            out.flush();
            System.out.println("send order to server succeed");
            String resp = in.readLine();
            System.out.println("Now is:" + resp);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("client finally");
            if(out != null) {
                out.close();
                out = null;
            }
            if(in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                in = null;
            }
            if(socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                socket = null;
            }
        }
    }
}

启动客户端，打印：

send order to server succeed
Now is:Wed Mar 25 18:25:45 CST 2020
client finally

服务端打印：

start a thread, hostName:127.0.0.1
start read data
server receive order:QUERY TIME ORDER
start read data
body is null

这样就完成了一次socket通信。

三、遇到的问题

3.1 server阻塞

问题现象：

客户端没有拿到返回值，通过Java VisualVM打印线程堆栈发现服务端阻塞在了body = in.readLine();

这里有个疑问，既然阻塞在readLine处为什么线程的状态还是RUNNABLE呢？这是因为这里的阻塞是IO阻塞，不是线程阻塞，这是两个概念，IO阻塞一般不会造成线程阻塞，至于IO阻塞中线程会不会占用CPU应该是有系统底层的线程调度决定，比如在Linux中等待IO的过程中线程不会占用CPU，知道IO完成会唤醒线程重新抢夺CPU时间片。

"Thread-0" #11 prio=5 os_prio=0 tid=0x0000000020d50000 nid=0x21f0 runnable [0x0000000021abe000]
   java.lang.Thread.State: RUNNABLE
        at java.net.SocketInputStream.socketRead0(Native Method)
        at java.net.SocketInputStream.socketRead(Unknown Source)
        at java.net.SocketInputStream.read(Unknown Source)
        at java.net.SocketInputStream.read(Unknown Source)
        at sun.nio.cs.StreamDecoder.readBytes(Unknown Source)
        at sun.nio.cs.StreamDecoder.implRead(Unknown Source)
        at sun.nio.cs.StreamDecoder.read(Unknown Source)
        - locked <0x00000006c2230df0> (a java.io.InputStreamReader)
        at java.io.InputStreamReader.read(Unknown Source)
        at java.io.BufferedReader.fill(Unknown Source)
        at java.io.BufferedReader.readLine(Unknown Source)
        - locked <0x00000006c2230df0> (a java.io.InputStreamReader)
        at java.io.BufferedReader.readLine(Unknown Source)
        at com.wangjun.io.bio.TimeServerHandler.run(TimeServerHandler.java:33)
        at java.lang.Thread.run(Unknown Source)

   Locked ownable synchronizers:
        - None

原因分析：

后来发现是因为客户端没有进行flush操作(最开始的时候忘记写out.flush();)，导致服务端阻塞在readline处。flush方法刷新输出流并强制将所有缓冲的输出字节被写出。

四、伪异步IO

为了解决一个线程维护一个连接的问题，我们可以通过线程池来维护连接，这样就可以控制线程的数量，灵活地调配线程资源，防止由于海量并发接入导致系统线程耗尽。通信模型如下：

4.1 代码实现

只需要将上面的TimerServer新建线程的地方改为线程池即可：

package com.wangjun.io.bio2;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 伪异步阻塞式IO创建时间服务器
 * 
 * @author wangjun
 * @date 2020-03-26
 * @version 1.0
 */
public class TimeServer {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // port采用默认值
            }
        }

        ServerSocket server = null;
        try {
            server = new ServerSocket(port);
            System.out.println("server started, port:" + port);
            Socket socket = null;
            // 创建线程池
            ExecutorService executor = Executors.newFixedThreadPool(20);
            while (true) {
                socket = server.accept();
                // 使用线程池处理请求
                System.out.println("use thread pool handler, hostName:" + socket.getInetAddress().getHostName());
                executor.submit(new TimeServerHandler(socket));
            }
        } finally {
            if (server != null) {
                System.out.println("server close");
                server.close();
                server = null;
            }
        }
    }
}

运行结果以上面一样。伪异步I/O实际上仅仅是对之前I/O线程模型的一个简单优化，它无法从根本上解决同步I/O导致的通信线程阻塞问题。

PreviousIO和NIO Next非阻塞式IO通信