java socket 同时读写_java 中的 BIO/NIO/AIO 详解

我们在前面学习了 linux 的 5 种 I/O 模型详解

下面我们一起来学习下如何使用 java 实现 BIO/NIO/AIO 这 3 种不同的网络 IO 模型编程。

BIO 作为最基础的 IO 版本，实现起来比较简单。

import java.io.BufferedReader;import java.io.IOException;import java.io.InputStreamReader;import java.io.PrintWriter;import java.net.ServerSocket;import java.net.Socket;/** * 

 BIO 服务端 

 * @author 老马啸西风 */public class TimeServer {    public static void main(String[] args) throws IOException {        final int port = 8088;        ServerSocket serverSocket = new ServerSocket(port);        System.out.println("server started at port " + port);        // 循环监听        while (true) {            Socket socket = serverSocket.accept();            System.out.println("客户端连接成功");            // 读取客户端的信息            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));            System.out.println("Server Recevie: " + bufferedReader.readLine());            // 读取客户端的信息            PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);            String currentTime = System.currentTimeMillis()+"";            printWriter.println(currentTime);        }    }}

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

 BIO 客户端 

 * * @author 老马啸西风 */public class TimeClient {    public static void main(String[] args) throws IOException {        final int port = 8088;        try(Socket clientSocket = new Socket("127.0.0.1", port)) {            System.out.println("Client started at port " + port);            // 写入信息            PrintWriter printWriter = new PrintWriter(clientSocket.getOutputStream(), true);            printWriter.println("hello bio");            // 读取反馈            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));            System.out.println("client recevie: " + bufferedReader.readLine());        }    }}

• 启动服务端

server started at port 8088

• 启动客户端

Client started at port 8088client recevie: 1568643464491Process finished with exit code 0

• 再次查看服务端日志

server started at port 8088客户端连接成功Server Recevie: hello bio

缺点其实非常明显，每次都要创建一个线程去处理。

比如我的实现是直接阻塞当前线程的，这当然非常的不友好。

可以使用线线程池的方式进行优化改进。

public class TimeThreadServer {    public static void main(String[] args) throws IOException {        final int port = 8088;        ServerSocket serverSocket = new ServerSocket(port);        System.out.println("server started at port " + port);        // 循环监听        while (true) {            Socket socket = serverSocket.accept();            System.out.println("客户端连接成功");            new ServerHandler(socket).start();        }    }    static class ServerHandler extends Thread {        private final Socket socket;        ServerHandler(Socket socket) {            this.socket = socket;        }        @Override        public void run() {            try {                // 读取客户端的信息                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));                System.out.println("Server Recevie: " + bufferedReader.readLine());                // 读取客户端的信息                PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);                String currentTime = System.currentTimeMillis()+"";                printWriter.println(currentTime);            } catch (IOException e) {                e.printStackTrace();            }        }    }}

public static void main(String[] args) throws IOException {    final int port = 8088;    ServerSocket serverSocket = new ServerSocket(port);    System.out.println("server started at port " + port);    ExecutorService executorService = Executors.newFixedThreadPool(2);    // 循环监听    while (true) {        Socket socket = serverSocket.accept();        System.out.println("客户端连接成功");        // 线程池处理        executorService.submit(new ServerHandler(socket));    }}

其他代码保持不变。

线程池版本的 BIO 又被称作伪异步 IO。

属于在 NIO 还没有流行之前的一种实战解决方案。

这种方式的性能和 BIO 想比较提升了很多，实现起来也比较简单，但是可靠性相对较差。

Java NIO Buffers用于和NIO Channel交互。正如你已经知道的，我们从channel中读取数据到buffers里，从buffer把数据写入到channels.

buffer 本质上就是一块内存区，可以用来写入数据，并在稍后读取出来。

这块内存被NIO Buffer包裹起来，对外提供一系列的读写方便开发的接口。

Java NIO Channel通道和流非常相似，主要有以下几点区别：

1. 通道可以读也可以写，流一般来说是单向的(只能读或者写)。
2. 通道可以异步读写。
3. 通道总是基于缓冲区Buffer来读写。

用单线程处理多个channels的好处是我需要更少的线程来处理channel。

实际上，你甚至可以用一个线程来处理所有的channels。

从操作系统的角度来看，切换线程开销是比较昂贵的，并且每个线程都需要占用系统资源，因此暂用线程越少越好。

需要留意的是，现代操作系统和CPU在多任务处理上已经变得越来越好，所以多线程带来的影响也越来越小。

如果一个CPU是多核的，如果不执行多任务反而是浪费了机器的性能。不过这些设计讨论是另外的话题了。

简而言之，通过Selector我们可以实现单线程操作多个channel。

NIO 采取通道(Channel)和缓冲区(Buffer)来传输和保存数据，它是非阻塞式的 I/O，即在等待连接、读写数据(这些都是在一线程以客户端的程序中会阻塞线程的操作)的时候，程序也可以做其他事情，以实现线程的异步操作。

考虑一个即时消息服务器，可能有上千个客户端同时连接到服务器，但是在任何时刻只有非常少量的消息需要读取和分发(如果采用线程池或者一线程一客户端方式，则会非常浪费资源)，这就需要一种方法能阻塞等待，直到有一个通道可以进行 I/O 操作。

NIO 的 Selector 选择器就实现了这样的功能，一个 Selector 实例可以同时检查一组信道的 I/O 状态，它就类似一个观察者，只要我们把需要探知的 SocketChannel 告诉 Selector,我们接着做别的事情，当有事件(比如，连接打开、数据到达等)发生时，它会通知我们，传回一组 SelectionKey,我们读取这些 Key,就会获得我们刚刚注册过的 SocketChannel,然后，我们从这个 Channel 中读取数据，接着我们可以处理这些数据。

Selector 内部原理实际是在做一个对所注册的 Channel 的轮询访问，不断的轮询(目前就这一个算法)，一旦轮询到一个 Channel 有所注册的事情发生，比如数据来了，它就会读取 Channel 中的数据，并对其进行处理。

要使用选择器，需要创建一个 Selector 实例，并将其注册到想要监控的信道上(通过 Channel 的方法实现)。

最后调用选择器的 select()方法，该方法会阻塞等待，直到有一个或多个信道准备好了 I/O 操作或等待超时，或另一个线程调用了该选择器的 wakeup()方法。

现在，在一个单独的线程中，通过调用 select()方法，就能检查多个信道是否准备好进行 I/O 操作，由于非阻塞 I/O 的异步特性，在检查的同时，我们也可以执行其他任务。

(1)创建一个 Selector 实例；

(2)将其注册到各种信道，并指定每个信道上感兴趣的I/O操作；

(3)重复执行：

调用一种 select() 方法；获取选取的键列表；对于已选键集中的每个键：    获取信道，并从键中获取附件(如果为信道及其相关的 key 添加了附件的话)；    确定准备就绪的操纵并执行，如果是 accept 操作，将接收的信道设置为非阻塞模式，并注册到选择器；    如果需要，修改键的兴趣操作集；    从已选键集中移除键。

import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.nio.channels.SocketChannel;import java.util.Iterator;import java.util.concurrent.TimeUnit;/** * @author binbin.hou * @since 1.0.0 */public class NioTcpServer {    /**     * 缓冲区的长度     */    private static final int BUFSIZE = 256;    /**     * select方法等待信道准备好的最长时间     */    private static final int TIMEOUT = 3000;    /**     * 监听的端口号     */    private static final int PORT = 18888;    public static void main(String[] args) throws IOException, InterruptedException {        // 1. 实例化一个通道        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();        // 设置为非阻塞模式        serverSocketChannel.configureBlocking(false);        // 绑定监听的端口        serverSocketChannel.socket().bind(new InetSocketAddress(PORT));        System.out.println("Server started listen on: " + PORT);        // 2. 构建一个 Selector，用于监听 Channel 的状态        Selector selector = Selector.open();        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);        //3. 不断循环等待        while (true) {            //3.1 循环等待直到有通道已经准备好            if(selector.select(TIMEOUT) == 0) {                System.out.println(".");                TimeUnit.SECONDS.sleep(1);                continue;            }            //3.2 遍历多有的 key            Iterator selectionKeySetIter = selector.selectedKeys().iterator();            while(selectionKeySetIter.hasNext()) {                SelectionKey selectionKey = selectionKeySetIter.next();                // accept I/O形式                if(selectionKey.isAcceptable()) {                    ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) selectionKey.channel();                    // 获取客户端 channel                    SocketChannel socketChannel = serverSocketChannel1.accept();                    socketChannel.configureBlocking(false);                    // 选择器注册监听的事件，同时制定关联的附件                    socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ | SelectionKey.OP_WRITE,                            ByteBuffer.allocate(BUFSIZE));                }                // 客户端信道已经准备好了读取数据到 buffer                if(selectionKey.isReadable()) {                    // 读取代码                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();                    // 获取对应的附件信息                    ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();                    long bufferRead = socketChannel.read(byteBuffer);                    //客户端关闭的链接。可以安全关闭                    if(bufferRead == -1) {                        socketChannel.close();                    } else {                        // 缓冲区读取到了数据，将其感兴趣的操作设置为可读可写。                        selectionKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);                        // 打印读取的内容                        System.out.println("Server read: " + new String(byteBuffer.array()));                    }                }                // 写入处理                if(selectionKey.isValid()                    && selectionKey.isWritable()) {                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();                    // 获取附件                    ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();                    // 重置缓冲区，准备将数据写入到信道                    byteBuffer.flip();                    socketChannel.write(byteBuffer);                    //Tells whether there are any elements between the current position and the limit.                    // 如果已经全部写入到信道，则将该信道感兴趣的操作标识为读                    if(!byteBuffer.hasRemaining()) {                        selectionKey.interestOps(SelectionKey.OP_READ);                    }                    // 为读取更多的数据腾出空间                    byteBuffer.compact();                }                // 手动删除                selectionKeySetIter.remove();            }        }    }}

import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.SocketChannel;import java.util.concurrent.TimeUnit;/** * @author binbin.hou * @since 1.0.0 */public class NioTcpClient {    /**     * 监听的端口号     */    private static final int PORT = 18888;    public static void main(String[] args) throws IOException, InterruptedException {        //1. 设置为非阻塞        SocketChannel socketChannel = SocketChannel.open();        socketChannel.configureBlocking(false);        socketChannel.connect(new InetSocketAddress(PORT));        //2. 连接中...        while (!socketChannel.finishConnect()) {            System.out.println(".");            TimeUnit.SECONDS.sleep(1);        }        System.out.println("");        //3. 写入/读取信息        String info = "hello nio test";        ByteBuffer readBuffer = ByteBuffer.allocate(info.length());        ByteBuffer writeBuffer = ByteBuffer.wrap(info.getBytes());        int totalReceivedBytes = 0;        int receivedBytes = 0;        while (totalReceivedBytes 

• 服务端

Server started listen on: 18888

• 客户端

Client received from server: hello nio test

• 服务端

...Server read: hello nio test                                                                                                               ...

jdk7中新增了一些与文件(网络)I/O相关的一些api。这些API被称为NIO.2，或称为AIO(Asynchronous I/O)。

AIO最大的一个特性就是异步能力，这种能力对socket与文件I/O都起作用。

• Future 方式

即提交一个 I/O 操作请求(accept/read/write)，返回一个 Future。

然后您可以对 Future 进行检查(调用get(timeout))，确定它是否完成，或者阻塞 IO 操作直到操作正常完成或者超时异常。

使用 Future 方式很简单，需要注意的是，因为Future.get()是同步的，所以如果不仔细考虑使用场合，使用 Future 方式可能很容易进入完全同步的编程模式，从而使得异步操作成为一个摆设。

如果这样，那么原来旧版本的 Socket API 便可以完全胜任，大可不必使用异步 I/O.

• Callback 方式

即提交一个 I/O 操作请求，并且指定一个 CompletionHandler。

当异步 I/O 操作完成时，便发送一个通知，此时这个 CompletionHandler 对象的 completed 或者 failed 方法将会被调用。

因为AIO的实施需充分调用OS参与，IO需要操作系统支持、并发也同样需要操作系统的支持，所以性能方面不同操作系统差异会比较明显。

import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.AsynchronousServerSocketChannel;import java.nio.channels.AsynchronousSocketChannel;import java.nio.charset.Charset;import java.util.concurrent.ExecutionException;import java.util.concurrent.Future;import java.util.concurrent.TimeUnit;import java.util.concurrent.TimeoutException;public class AioFutureServer {    private static final int DEFAULT_PORT = 12345;    private AsynchronousServerSocketChannel serverSocketChannel;    public AioFutureServer() throws IOException {        serverSocketChannel = AsynchronousServerSocketChannel.open();        serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT));        System.out.println("Server listen on port: " + DEFAULT_PORT);    }    public void startWithFuture() throws InterruptedException,            ExecutionException, TimeoutException {        while (true) {            // 循环接收客户端请求            Future future = serverSocketChannel.accept();            // get() 是为了确保 accept 到一个连接            AsynchronousSocketChannel socket = future.get();            handleWithFuture(socket);        }    }    /**     * 处理未来的信息     * @param channel 异步客户端     */    private void handleWithFuture(AsynchronousSocketChannel channel) throws InterruptedException, ExecutionException, TimeoutException {        ByteBuffer readBuf = ByteBuffer.allocate(8);        readBuf.clear();        // 一次可能读不完        while (true) {            //get 是为了确保 read 完成，超时时间可以有效避免DOS攻击，如果客户端一直不发送数据，则进行超时处理            Integer integer = channel.read(readBuf).get(10, TimeUnit.SECONDS);            System.out.println("read: " + integer);            if (integer == -1) {                break;            }            readBuf.flip();            System.out.println("received: " + Charset.forName("UTF-8").decode(readBuf));            readBuf.clear();        }    }    public static void main(String[] args) throws IOException, InterruptedException, ExecutionException, TimeoutException {        new AioFutureServer().startWithFuture();    }}

import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.AsynchronousServerSocketChannel;import java.nio.channels.AsynchronousSocketChannel;import java.nio.charset.Charset;import java.util.concurrent.ExecutionException;import java.util.concurrent.Future;import java.util.concurrent.TimeUnit;import java.util.concurrent.TimeoutException;public class AioClient {    private static final int DEFAULT_PORT = 12345;    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {        AsynchronousSocketChannel client = AsynchronousSocketChannel.open();        client.connect(new InetSocketAddress("localhost", DEFAULT_PORT)).get();        client.write(ByteBuffer.wrap("123456789".getBytes()));    }}

• 启动服务端

Server listen on port: 12345

• 启动客户端

服务端日志

read: 8received: 12345678read: 1received: 9Exception in thread "main" java.util.concurrent.ExecutionException: java.io.IOException: 指定的网络名不再可用。

import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.AsynchronousServerSocketChannel;import java.nio.channels.AsynchronousSocketChannel;import java.nio.channels.CompletionHandler;import java.nio.charset.Charset;import java.util.concurrent.TimeUnit;public class AioCompletionServer {    private static final int DEFAULT_PORT = 12345;    private AsynchronousServerSocketChannel serverSocketChannel;    public AioCompletionServer() throws IOException {        serverSocketChannel = AsynchronousServerSocketChannel.open();        serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT));        System.out.println("Server listen on port: " + DEFAULT_PORT);    }    /**     * 使用回调的方式     */    public void startWithCompletionHandler() {        serverSocketChannel.accept(null,                new CompletionHandler() {                    @Override                    public void completed(AsynchronousSocketChannel result, Object attachment) {                        // 再此接收客户端连接                        serverSocketChannel.accept(null, this);                        // 处理结果                        handleWithCompletionHandler(result);                    }                    @Override                    public void failed(Throwable exc, Object attachment) {                        exc.printStackTrace();                    }                });    }    /**     * 处理异步的结果     * @param channel 客户端信道     */    private void handleWithCompletionHandler(final AsynchronousSocketChannel channel) {        try {            final long timeout = 10L;            final ByteBuffer buffer = ByteBuffer.allocate(8);            // 再次读取，还是一种回调的方式。            channel.read(buffer, timeout, TimeUnit.SECONDS, null, new CompletionHandler() {                @Override                public void completed(Integer result, Object attachment) {                    System.out.println("read:" + result);                    if (result == -1) {                        try {                            channel.close();                        } catch (IOException e) {                            e.printStackTrace();                        }                        return;                    }                    buffer.flip();                    System.out.println("received message:" + Charset.forName("UTF-8").decode(buffer));                    buffer.clear();                    // 递归调用，直到结束为止。                    channel.read(buffer, timeout, TimeUnit.SECONDS, null, this);                }                @Override                public void failed(Throwable exc, Object attachment) {                    exc.printStackTrace();                }            });        } catch (Exception e) {            e.printStackTrace();        }    }    public static void main(String[] args) throws IOException, InterruptedException {        new AioCompletionServer().startWithCompletionHandler();        // 沉睡等待处理。        TimeUnit.SECONDS.sleep(100);    }}

同上

本文讲述了 jdk 实现的 bio/nio/aio 的方式，你是否会感觉 jdk 中的 api 设计过于复杂呢？

下一节我们将通过 netty 框架实现上述功能，并讲述我们为什么要选择 netty 作为网络开发的基本工具。

希望本文对你有所帮助，如果喜欢，欢迎点赞收藏转发一波。

我是老马，期待与你的下次相遇。