背景

最近在回顾一下Java IO相关的知识，顺带写一下入门级别的文章。感觉工作以后很少写文章，一直想写点高质量的文章导致最后一篇文章都很难写。所以不写原理，只写实践，随大流，有问题请留言。(后续有时间再补充原理性的东西，从硬件到操作系统到JVM到JDK)

实现案例

创建一个server，可以接受多个client端的连接，接收到信息后返回一个接收到的信息。

传统IO实现

传统的IO就是我们所说的BIO（block io），

server端源码如下

package tech.sohocoder.postman.io;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

public class Server {

    private ServerSocket serverSocket;


    private void start() throws IOException, ClassNotFoundException {
        InetSocketAddress inetSocketAddress = new InetSocketAddress("localhost", 9000);
        serverSocket = new ServerSocket();
        serverSocket.bind(inetSocketAddress);
        ExecutorService executorService = Executors.newCachedThreadPool(new CaughtExceptionsThreadFactory());
        while (true) {
            Socket socket = serverSocket.accept();
            System.out.println("accept socket: " + socket.getRemoteSocketAddress());
            executorService.submit(new SocketHandler(socket));
        }
    }

    private static class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
        @Override public void uncaughtException(Thread t, Throwable e) {
            e.printStackTrace();
        }
    }

    private class SocketHandler implements Runnable {

        private Socket socket;

        public SocketHandler(Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
                    String message = ois.readObject().toString();
                    System.out.println("Message Received: " + message);
                    ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
                    //write object to Socket
                    oos.writeObject("Hi Client " + message);
                    if (message.equals("quit")) {
                        ois.close();
                        oos.close();
                        socket.close();
                        break;
                    }
                }
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException {
        Server server = new Server();
        server.start();
    }
}

client端源码如下

package tech.sohocoder.postman.io;

import java.io.*;
import java.net.Socket;

public class Client {

    private Socket socket;

    public void start() throws IOException, ClassNotFoundException {
        socket = new Socket("localhost", 9000);
        if(socket.isConnected()) {
            System.out.println("socket is connected");
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
            for (;;) {
                final String input = in.readLine();
                final String line = input != null ? input.trim() : null;
                if (null == line || line.isEmpty()) { // skip `enter` or `enter` with spaces.
                    continue;
                }
                ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
                oos.writeObject(line);
                ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
                System.out.println("Message: " + ois.readObject());
                if(line.equals("quit")) {
                    oos.close();
                    ois.close();
                    socket.close();
                    break;
                }
            }
        }
        System.out.println("Bye");
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException {
        Client client = new Client();
        client.start();
    }
}

NIO的阻塞实现

NIO实际上就是面向缓存及通道的新型IO(由JSR 51定义，后面JSR 203进行了扩展，有兴趣阅读一下这两个JSR)可以支持阻塞和非阻塞方式。先实现一下阻塞方式

client

package tech.sohocoder.nio.block;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

import static java.lang.System.out;

public class Client {

    private SocketChannel socketChannel;

    public void start() throws IOException {
        socketChannel = SocketChannel.open();
        SocketAddress socketAddress = new InetSocketAddress("localhost", 9000);
        socketChannel.connect(socketAddress);

        BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
        for (;;) {
            final String input = in.readLine();
            final String line = input != null ? input.trim() : null;
            if (null == line || line.isEmpty()) { // skip `enter` or `enter` with spaces.
                continue;
            }

            ByteBuffer byteBuffer = ByteBuffer.wrap(line.getBytes());
            socketChannel.write(byteBuffer);

            if(line.equals("quit")) {
                out.println("quit!");
                socketChannel.close();
                break;
            }

            ByteBuffer returnByteBuffer = ByteBuffer.allocate(1024);
            socketChannel.read(returnByteBuffer);
            String message = new String(returnByteBuffer.array()).trim();
            out.println("Receive message: " + message);
        }
    }

    public static void main(String[] args) throws IOException {
        tech.sohocoder.nio.noblock.Client client = new tech.sohocoder.nio.noblock.Client();
        client.start();
    }
}

server

package tech.sohocoder.nio.block;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class Server {

    private ServerSocketChannel serverSocketChannel;

    private void start() throws IOException {
        serverSocketChannel = ServerSocketChannel.open();
        SocketAddress socketAddress = new InetSocketAddress(9000);
        serverSocketChannel.bind(socketAddress);

        while (true) {
            System.out.println("listening...");
            SocketChannel socketChannel = serverSocketChannel.accept();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            int readLength = socketChannel.read(byteBuffer);
            if(readLength != -1) {
                String receiveStr = new String(byteBuffer.array()).trim();
                System.out.println(receiveStr);
                socketChannel.write(byteBuffer);
            }
            socketChannel.close();
        }
    }

    public static void main(String[] args) throws IOException {
        Server server = new Server();
        server.start();
    }
}

NIO的非阻塞方式

NIO如果需要非阻塞，需要使用到selector。selector是在JDK1.4加入，主要是用于支持IO多路复用，Linux下jdk实现就是基于epoll。

client端代码保存一致。

server端实际上就是使用一个线程来支持多个连接

package tech.sohocoder.nio.noblock;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
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.Set;

import static java.lang.System.out;

public class Server {

    private ServerSocketChannel serverSocketChannel;

    private Selector selector;

    private void start() throws IOException, InterruptedException {
        serverSocketChannel = ServerSocketChannel.open();
        SocketAddress socketAddress = new InetSocketAddress(9000);
        serverSocketChannel.bind(socketAddress);

        serverSocketChannel.configureBlocking(false);

        int opSelectionKey = serverSocketChannel.validOps();

        selector = Selector.open();

        SelectionKey selectionKey = serverSocketChannel.register(selector, opSelectionKey);

        out.println(selector);
        out.println(selectionKey);
        while(true) {
            out.println("waiting for connected...");
            selector.select();
            Set<SelectionKey> set  = selector.selectedKeys();
            Iterator<SelectionKey> iterator = set.iterator();
            while (iterator.hasNext()) {
                SelectionKey mySelectionKey = iterator.next();
                if(mySelectionKey.isAcceptable()) {
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    SelectionKey selectionKey1 = socketChannel.register(selector, SelectionKey.OP_READ);
                    out.println("socket channel selectionkey: " + selectionKey1);
                    out.println("connect from : " + socketChannel.getRemoteAddress());
                }else if(mySelectionKey.isReadable()){
                    SocketChannel socketChannel = (SocketChannel) mySelectionKey.channel();
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    socketChannel.read(byteBuffer);
                    String message = new String(byteBuffer.array()).trim();
                    out.println("Receive message: " + message);
                    if(message.equals("quit")) {
                        out.println("close connection: " + socketChannel.getRemoteAddress());
                        socketChannel.close();
                        mySelectionKey.cancel();
                    }else {
                        ByteBuffer returnByteBuffer = ByteBuffer.wrap(" receive your message".getBytes());
                        socketChannel.write(returnByteBuffer);
                    }
                }
                iterator.remove();
            }
        }

    }

    public static void main(String[] args) throws IOException, InterruptedException {
        Server server = new Server();
        server.start();
    }
}

AIO实现

上面的IO，NIO的阻塞实际上是同步阻塞的方式，NIO的非阻塞是同步非阻塞方式。AIO(asynchronous I/O))是异步IO，实现是异步非阻塞方式，在jdk1.7中引入。

server端源码如下：

package tech.sohocoder.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static java.lang.System.out;

public class Server {

    private AsynchronousServerSocketChannel asynchronousServerSocketChannel;

    private void start() throws IOException, InterruptedException {
        // worker thread pool
        AsynchronousChannelGroup asynchronousChannelGroup = AsynchronousChannelGroup.withCachedThreadPool(Executors.newCachedThreadPool(), 4);
        asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open(asynchronousChannelGroup);
        int port = 9000;
        InetSocketAddress socketAddress = new InetSocketAddress("localhost", port);
        asynchronousServerSocketChannel.bind(socketAddress);

        out.println("Starting listening on port " + port);
        // add handler
        asynchronousServerSocketChannel.accept(asynchronousServerSocketChannel, new CompletionHandler<AsynchronousSocketChannel, Object>() {
            @Override
            public void completed(AsynchronousSocketChannel asynchronousSocketChannel, Object o) {

                try {
                    out.println("connect from : " + asynchronousSocketChannel.getRemoteAddress());
                } catch (IOException e) {
                    e.printStackTrace();
                }
                // accept next connection
                asynchronousServerSocketChannel.accept(asynchronousServerSocketChannel, this);
                while (true) {
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    Future<Integer> future = asynchronousSocketChannel.read(byteBuffer);
                    try {
                        future.get();
                        String message = new String(byteBuffer.array()).trim();
                        out.println("Receive message: " + message);
                        if (message.equals("quit")) {
                            out.println("close client: " + asynchronousSocketChannel.getRemoteAddress());
                            asynchronousSocketChannel.close();
                            break;
                        }

                        ByteBuffer returnByteBuffer = ByteBuffer.wrap("receive your message".getBytes());
                        Future<Integer> returnFuture = asynchronousSocketChannel.write(returnByteBuffer);
                        returnFuture.get();

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

            @Override
            public void failed(Throwable throwable, Object o) {
                out.println("error to accept: " + throwable.getMessage());
            }
        });
        asynchronousChannelGroup.awaitTermination(Integer.MAX_VALUE, TimeUnit.MINUTES);
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        Server server = new Server();
        server.start();
    }
}

Netty实现

Netty是java中使用很广泛的库，既可以实现NIO也可以实现AIO，还是针对上面的例子来实现一下

server端

package tech.sohocoder.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import static java.lang.System.out;

public class Server {

    private void start() throws InterruptedException {
        EventLoopGroup bossEventLoopGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerEventLoopGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossEventLoopGroup, workerEventLoopGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // add handler into pipeline
                            socketChannel.pipeline()
                                    .addLast(new StringDecoder())
                                    .addLast(new StringEncoder())
                                    .addLast(new ServerHandler());
                        }
                    });

            ChannelFuture channelFuture = serverBootstrap.bind(9000).sync();
            out.println("listening...");
            channelFuture.channel().closeFuture().sync();
        }finally {
           bossEventLoopGroup.shutdownGracefully();
        }

    }

    public static void main(String[] args) throws InterruptedException {
        Server server = new Server();
        server.start();
    }

}

这里面需要使用到ServerHandler，具体代码如下

package tech.sohocoder.netty;

import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;

import java.net.SocketAddress;

import static java.lang.System.out;

public class ServerHandler extends ChannelDuplexHandler {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        out.println("Receive message: " + msg);
        String message = "receive your message";
        ctx.writeAndFlush(message);
    }

    @Override
    public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
        super.connect(ctx, remoteAddress, localAddress, promise);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        out.println("connect from: " + ctx.channel().remoteAddress().toString());
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        out.println("close connection: " + ctx.channel().remoteAddress().toString());
        super.channelInactive(ctx);
    }

}

client端也用netty写一下

package tech.sohocoder.aio;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

import static java.lang.System.out;

public class Client {

    private SocketChannel socketChannel;

    public void start() throws IOException {
        socketChannel = SocketChannel.open();
        SocketAddress socketAddress = new InetSocketAddress("localhost", 9000);
        socketChannel.connect(socketAddress);
        if(socketChannel.isConnected()) {
            out.println("connect to " + socketChannel.getRemoteAddress());
        }

        BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
        for (;;) {
            final String input = in.readLine();
            final String line = input != null ? input.trim() : null;
            if (null == line || line.isEmpty()) { // skip `enter` or `enter` with spaces.
                continue;
            }

            ByteBuffer byteBuffer = ByteBuffer.wrap(line.getBytes());
            socketChannel.write(byteBuffer);

            if(line.equals("quit")) {
                out.println("quit!");
                socketChannel.close();
                break;
            }

            ByteBuffer returnByteBuffer = ByteBuffer.allocate(1024);
            socketChannel.read(returnByteBuffer);
            String message = new String(returnByteBuffer.array()).trim();
            out.println("Receive message: " + message);
        }
    }

    public static void main(String[] args) throws IOException {
        Client client = new Client();
        client.start();
    }
}

同样要实现一个ClientHandler

package tech.sohocoder.netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import static java.lang.System.out;

public class ClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        out.println("Receive message: " + s);
    }
}