必须明白的几个概念

阻塞(Block)和非阻塞(Non-Block)

阻塞和非阻塞是在缓冲区数据没有准备就绪的情况下处理方式
阻塞:在没有就绪下,就一直等待哪里
非阻塞:在没有就绪下,则直接返回
阻塞和非阻塞是相对于io操作来讲
举例如下图:从机器内存到jvm内存

image.png

同步(Synchronization)和异步(Asynchronous)

同步和异步是处理io事件的采用方式
同步:应用程序参与io的处理
异步:io交给操作系统,应用程序只需要等待通知
异步和同步是相对于时间点,异步指同一时间点做多个处理,同步指同一时间点做一个处理。

读和写

读指input,写指output,输入和输出是相对于内存来讲的

BIO 与 NIO 对比

BIO:面向流(乡村公路),阻塞 IO(多线程)
NIO:面向缓冲(高速公路，多路复用技术),非阻塞IO(反应堆 Reactor),选择器(轮询机制)

image.png

阻塞和非阻塞

NIO是发现缓冲数据没有准备好,直接返回,处理其他事情,因此可以一个线程处理多个输入和输出通道。
BIO 是发现缓冲区数据没有准备好,则等待哪里,因此,可以一个线程处理一个输入和输出通道。

Java AIO

真正实现异步IO

代码实现

BIO

1 服务端

//同步阻塞IO模型
public class BIOServer {
    ServerSocket server;
    public BIOServer(int port){
        try {
            server = new ServerSocket(port);
            System.out.println("BIO服务已启动，监听端口是：" + port);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void listen() throws IOException{
        //循环监听
        while(true){
            //等待客户端连接，阻塞方法
            Socket client = server.accept();

            InputStream is = client.getInputStream();
            //网络客户端把数据发送到网卡，机器所得到的数据读到了JVM内中
            byte [] buff = new byte[1024];
            int len = is.read(buff);
            if(len > 0){
                String msg = new String(buff,0,len);
                System.out.println("收到" + msg);
            }
        }
    }   
    public static void main(String[] args) throws IOException {
        new BIOServer(8080).listen();
    }
}

2 客户端

public class BIOClient {
    public static void main(String[] args) throws UnknownHostException, IOException {
        Socket client = new Socket("localhost", 8080);

        OutputStream os = client.getOutputStream();

        //生成一个随机的ID
        String name = UUID.randomUUID().toString();

        System.out.println("客户端发送数据：" + name);
        os.write(name.getBytes());
        os.close();
        client.close();
    }
}

NIO

1 服务端

/**
 * NIO的操作过于繁琐，于是才有了Netty
 * Netty就是对这一系列非常繁琐的操作进行了封装
 */
public class NIOServerDemo {

    private int port = 8080;

    //轮询器 Selector 大堂经理
    private Selector selector;
    //缓冲区 Buffer 等候区
    private ByteBuffer buffer = ByteBuffer.allocate(1024);

    //初始化完毕
    public NIOServerDemo(int port){
        try {
            this.port = port;

            //远程服务器通道
            ServerSocketChannel server = ServerSocketChannel.open();
            //我得告诉地址
            server.bind(new InetSocketAddress(this.port));
            //BIO 升级版本 NIO，为了兼容BIO，NIO模型默认是采用阻塞式
            server.configureBlocking(false);

            //轮训器
            selector = Selector.open();

            //轮训器属于该通道
            server.register(selector, SelectionKey.OP_ACCEPT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void listen(){
        System.out.println("listen on " + this.port + ".");
        try {
            //轮询主线程
            while (true){
                //通道个数,当没有通道则阻塞,有则继续进行
                selector.select();
                //每次都拿到所有的号子
                Set<SelectionKey> keys = selector.selectedKeys();
                Iterator<SelectionKey> iter = keys.iterator();
                //不断地迭代，就叫轮询
                //同步体现在这里，因为每次只能拿一个key，每次只能处理一种状态
                while (iter.hasNext()){
                    SelectionKey key = iter.next();
                    iter.remove();
                    //每一个key代表一种状态
                    //数据就绪、数据可读、数据可写 等等等等
                    process(key);
                }
                
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //具体办业务的方法，坐班柜员
    //在同一时间点，只能干一件事
    private void process(SelectionKey key) throws IOException {
        //针对于每一种状态给一个反应
        if(key.isAcceptable()){
            ServerSocketChannel server = (ServerSocketChannel)key.channel();
            //这个方法体现非阻塞，不管你数据有没有准备好
            //你给我一个状态和反馈
            SocketChannel channel = server.accept();
            //一定一定要记得设置为非阻塞
            channel.configureBlocking(false);
            //当数据准备就绪的时候，将状态改为可读
            channel.register(selector,SelectionKey.OP_READ);
        }
        else if(key.isReadable()){
            //key.channel 从多路复用器中拿到客户端的引用
            SocketChannel channel = (SocketChannel)key.channel();
            int len = channel.read(buffer);
            if(len > 0){
                buffer.flip();
                String content = new String(buffer.array(),0,len);
                key = channel.register(selector,SelectionKey.OP_WRITE);
                //在key上携带一个附件，一会再写出去
                key.attach(content);
                System.out.println("读取内容：" + content);
            }
        }
        else if(key.isWritable()){
            SocketChannel channel = (SocketChannel)key.channel();

            String content = (String)key.attachment();
            channel.write(ByteBuffer.wrap(("输出：" + content).getBytes()));

            channel.close();
        }
    }
    public static void main(String[] args) {
        new NIOServerDemo(8080).listen();
    }
}

2 客户端
同BIO客户端

AIO

1 服务端

public class AIOServer {
    private final int port;
    public static void main(String args[]) {
        int port = 8000;
        new AIOServer(port);
    }
    public AIOServer(int port) {
        this.port = port;
        listen();
    }
    private void listen() {
        try {
            ExecutorService executorService = Executors.newCachedThreadPool();
            AsynchronousChannelGroup threadGroup = AsynchronousChannelGroup.withCachedThreadPool(executorService, 1);
            //开门营业
            //工作线程，用来侦听回调的，事件响应的时候需要回调
            final AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open(threadGroup);
            server.bind(new InetSocketAddress(port));
            System.out.println("服务已启动，监听端口" + port);

            //准备接受数据
            server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>(){
                final ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
                //实现completed方法来回调
                //由操作系统来触发
                //回调有两个状态，成功
                public void completed(AsynchronousSocketChannel result, Object attachment){
                    System.out.println("IO操作成功，开始获取数据");
                    try {
                        buffer.clear();
                        result.read(buffer).get();
                        buffer.flip();
                        result.write(buffer);
                        buffer.flip();
                    } catch (Exception e) {
                        System.out.println(e.toString());
                    } finally {
                        try {
                            result.close();
                            server.accept(null, this);
                        } catch (Exception e) {
                            System.out.println(e.toString());
                        }
                    }

                    System.out.println("操作完成");
                }

                @Override
                //回调有两个状态，失败
                public void failed(Throwable exc, Object attachment) {
                    System.out.println("IO操作是失败: " + exc);
                }
            });

            try {
                Thread.sleep(Integer.MAX_VALUE);
            } catch (InterruptedException ex) {
                System.out.println(ex);
            }
        } catch (IOException e) {
            System.out.println(e);
        }
    }
}

2 客户端

public class AIOClient {
    private final AsynchronousSocketChannel client;

    public AIOClient() throws Exception{
        client = AsynchronousSocketChannel.open();
    }

    public void connect(String host,int port)throws Exception{
        client.connect(new InetSocketAddress(host,port),null,new CompletionHandler<Void,Void>() {
            @Override
            public void completed(Void result, Void attachment) {
                try {
                    client.write(ByteBuffer.wrap("这是一条测试数据".getBytes())).get();
                    System.out.println("已发送至服务器");
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }

            @Override
            public void failed(Throwable exc, Void attachment) {
                exc.printStackTrace();
            }
        });
        final ByteBuffer bb = ByteBuffer.allocate(1024);
        client.read(bb, null, new CompletionHandler<Integer,Object>(){

                    @Override
                    public void completed(Integer result, Object attachment) {
                        System.out.println("IO操作完成" + result);
                        System.out.println("获取反馈结果" + new String(bb.array()));
                    }

                    @Override
                    public void failed(Throwable exc, Object attachment) {
                        exc.printStackTrace();
                    }
                }
        );

        try {
            Thread.sleep(Integer.MAX_VALUE);
        } catch (InterruptedException ex) {
            System.out.println(ex);
        }

    }
    public static void main(String args[])throws Exception{
        new AIOClient().connect("localhost",8000);
    }
}