IO内存以及Reactor模型(更新完了)

待我称王封你为后i 2023-02-28 05:30 6阅读 0赞

### 文章目录 ###

*  同步IO
 *  select模型/poll模型
 *  epoll模型
 *  I/O线程模型
 *   *  Reactor模型
     *   *  单线程
         *   *  下面附赠我写的可跑的案例
         *  多线程
         *  主从模式
         *   *  下面附赠我写的可跑的案例
     *  Proactor模型

[参考C实现,select、poll、epoll、以及反应堆案例l][C_select_poll_epoll_l]

# 同步IO #

对文件设备数据的读写需要同步等待操作系统内核，即使文件设备并没有数据可读，线程也会被阻塞住，虽然阻塞时不占用CPU始终周期，但是若需要支持并发连接，则必须启用大量的线程，即每个连接一个线程。这样必不可少的会造成线程大量的上下文切换，随着并发量的增高，性能越来越差。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70]

# select模型/poll模型 #

*  通过一个线程不断的判断文件句柄数组是否有准备就绪的文件设备，这样就不需要每个线程同步等待，减少了大量线程，降低了线程上下文切换带来的性能损失，提高了线程利用率。这种方式也称为I/O多路复用技术。
 *  select模型需要对数组进行遍历，因此时间复杂度是O(n)因此当高并发量的时候，select模型性能会越来越差。而且受限于文件描述符限制,32位1024,64为2048.
 *  poll模型和select差不多,只是浅浅的封装了下,我都怀疑作者是不是好玩,但使用的是链表存储,因此不会受限于数组容量限制,解决了并发限制问题,但依旧存在大量连接,遍历查询是否有准备就绪的问题.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 1]

# epoll模型 #

在linux2.6支持了epoll模型，epoll模型解决了select模型的性能瓶颈问题。它通过注册回调事件的方式，当数据可读写时，将其加入到一个可读写事件的队列中。这样每次用户获取时不需要遍历所有句柄，时间复杂度降低为O(1)。因此epoll不会随着并发量的增加而性能降低。随着epoll模型的出现C10K的问题已经完美解决。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center]

# I/O线程模型 #

[摘自Doug Lea大神的一篇文章][Doug Lea]

从线程模型上常见的线程模型有Reactor模型和Proactor模型，无论是哪种线程模型都使用I/O多路复用技术，使用一个线程将I/O读写操作转变为读写事件，我们将这个线程称之为多路分离器。

对应上I/O模型，Reacor模型属于同步I/O模型，Proactor模型属于异步I/O模型。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 1]

class Server implements Runnable {
        
        public void run() {
        
            try {
        
                ServerSocket ss = new ServerSocket(PORT);
                while (!Thread.interrupted())
                new Thread(new Handler(ss.accept())).start(); //创建新线程来handle
                // or, single-threaded, or a thread pool
            } catch (IOException ex) {
         /* ... */ }
        }
        
        static class Handler implements Runnable {
        
            final Socket socket;
            Handler(Socket s) {
         socket = s; }
            public void run() {
        
                try {
        
                    byte[] input = new byte[MAX_INPUT];
                    socket.getInputStream().read(input);
                    byte[] output = process(input);
                    socket.getOutputStream().write(output);
                } catch (IOException ex) {
         /* ... */ }
            }       
            private byte[] process(byte[] cmd) {
         /* ... */ }
        }
    }

## Reactor模型 ##

在Reactor中，需要先注册事件就绪事件，网卡接收到数据时，DMA将数据从网卡缓冲区传输到内核缓冲区时，就会通知多路分离器读事件就绪，此时我们需要从内核空间读取到用户空间。

同步I/O采用缓冲I/O的方式，首先内核会从申请一个内存空间用于存放输入或输出缓冲区，数据都会先缓存在该缓冲区。

### 单线程 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 2]

class Reactor implements Runnable {
         
        final Selector selector;
        final ServerSocketChannel serverSocket;
        Reactor(int port) throws IOException {
         //Reactor初始化
            selector = Selector.open();
            serverSocket = ServerSocketChannel.open();
            serverSocket.socket().bind(new InetSocketAddress(port));
            serverSocket.configureBlocking(false); //非阻塞
    //        与Selector一起使用时，Channel必须处于非阻塞模式下。
            SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT); //分步处理,第一步,接收accept事件
            sk.attach(new Acceptor()); //attach callback object, Acceptor
        }
        
        public void run() {
         
            try {
        
                while (!Thread.interrupted()) {
        
                    selector.select();
                    Set selected = selector.selectedKeys();
                    Iterator it = selected.iterator();
                    while (it.hasNext())
                        dispatch((SelectionKey)(it.next()); //Reactor负责dispatch收到的事件
                    selected.clear();
                }
            } catch (IOException ex) {
         /* ... */ }
        }
        
        void dispatch(SelectionKey k) {
        
            Runnable r = (Runnable)(k.attachment()); //调用之前注册的callback对象
            if (r != null)
                r.run();
        }
        
        class Acceptor implements Runnable {
         // inner
            public void run() {
        
                try {
        
                    SocketChannel c = serverSocket.accept();
                    if (c != null)
                    new Handler(selector, c);
                }
                catch(IOException ex) {
         /* ... */ }
            }
        }
    }
    
    final class Handler implements Runnable {
        
        final SocketChannel socket;
        final SelectionKey sk;
        ByteBuffer input = ByteBuffer.allocate(MAXIN);
        ByteBuffer output = ByteBuffer.allocate(MAXOUT);
        static final int READING = 0, SENDING = 1;
        int state = READING;
        
        Handler(Selector sel, SocketChannel c) throws IOException {
        
            socket = c; 
            //设置非阻塞，读取立即返回
            c.configureBlocking(false);
            // Optionally try first read now
            sk = socket.register(sel, 0);
            sk.attach(this); //将Handler作为callback对象
            sk.interestOps(SelectionKey.OP_READ); //第二步,接收Read事件
    //某个线程调用select()方法后阻塞了，即使没有通道已经就绪，也有办法让其从select()方法返回。
    //只要让其它线程在第一个线程调用select()方法的那个对象上调用Selector.wakeup()方法即可。
    //阻塞在select()方法上的线程会立马返回。
            //如果有其它线程调用了wakeup()方法，但当前没有线程阻塞在select()方法上，
            //下个调用select()方法的线程会立即“醒来（wake up）”。
            sel.wakeup();
        }
        boolean inputIsComplete() {
         /* ... */ }
        boolean outputIsComplete() {
         /* ... */ }
        void process() {
         /* ... */ }
        
        public void run() {
        
            try {
        
                if (state == READING) read();
                else if (state == SENDING) send();
            } catch (IOException ex) {
         /* ... */ }
        }
        
        void read() throws IOException {
        
            socket.read(input);
            if (inputIsComplete()) {
        
                process();
                state = SENDING;
                // Normally also do first write now
                sk.interestOps(SelectionKey.OP_WRITE); //第三步,接收write事件
            }
        }
        void send() throws IOException {
        
            socket.write(output);
            if (outputIsComplete()) sk.cancel(); //write完就结束了, 关闭select key
        }
    }
    
    //上面 的实现用Handler来同时处理Read和Write事件, 所以里面出现状态判断
    //我们可以用State-Object pattern来更优雅的实现
    class Handler {
         // ...
        public void run() {
         // initial state is reader
            socket.read(input);
            if (inputIsComplete()) {
        
                process();
                sk.attach(new Sender());  //状态迁移, Read后变成write, 用Sender作为新的callback对象
                sk.interest(SelectionKey.OP_WRITE);
                sk.selector().wakeup();
            }
        }
        class Sender implements Runnable {
        
            public void run(){
         // ...
                socket.write(output);
                if (outputIsComplete()) sk.cancel();
            }
        }
    }

可以发现handler和accept在同一个线程,假设当某个handler阻塞,则会阻塞所有的请求.  
解决办法,将handler使用线程池管理.

#### 下面附赠我写的可跑的案例 ####

因为最近无聊，考虑手写一个netty。

package netty.reactor.reactor;
    
    import netty.reactor.Config;
    
    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.Set;
    
    /**
     * Reactor线程模型
     * 单线程
     *
     * @author WangChao
     * @create 2021/5/16 22:51
     */
    public class Reactor implements Runnable {
        
        public static void main(String[] args) throws IOException {
        
            new Reactor().run();
        }
    
        final Selector selector;
        final ServerSocketChannel serverSocket;
    
        Reactor() throws IOException {
        
            //Reactor初始化
            selector = Selector.open();
            serverSocket = ServerSocketChannel.open();
            serverSocket.socket().bind(new InetSocketAddress(Config.HOST, Config.PORT));
            //非阻塞
            serverSocket.configureBlocking(false);
            //与Selector一起使用时，Channel必须处于非阻塞模式下。
            SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
            //分步处理,第一步,接收accept事件
            //attach callback object, Acceptor
            sk.attach(new Acceptor());
        }
    
        @Override
        public void run() {
        
            try {
        
                while (!Thread.interrupted()) {
        
                    selector.select();
                    Set selected = selector.selectedKeys();
                    Iterator it = selected.iterator();
                    while (it.hasNext()) {
        
                        //Reactor负责dispatch收到的事件
                        dispatch((SelectionKey) (it.next()));
                    }
                    selected.clear();
                }
            } catch (IOException ex) {
         /* ... */ }
        }
    
        void dispatch(SelectionKey k) {
        
            //调用之前注册的callback对象
            Runnable r = (Runnable) (k.attachment());
            if (r != null) {
        
                r.run();
            }
        }
    
        class Acceptor implements Runnable {
         // inner
            @Override
            public void run() {
        
                try {
        
                    SocketChannel c = serverSocket.accept();
                    if (c != null) {
        
                        new Handler(selector, c);
                    }
                } catch (IOException ex) {
         /* ... */ }
            }
        }
    }
    
    final class Handler implements Runnable {
        
        final SocketChannel socket;
        final SelectionKey sk;
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        static final int READING = 0, SENDING = 1;
        int state = READING;
    
        Handler(Selector sel, SocketChannel c) throws IOException {
        
            socket = c;
            //设置非阻塞，读取立即返回
            c.configureBlocking(false);
            // Optionally try first read now
            sk = socket.register(sel, 0);
            //将Handler作为callback对象
            sk.attach(this);
            //第二步,接收Read事件
            sk.interestOps(SelectionKey.OP_READ);
            //某个线程调用select()方法后阻塞了，即使没有通道已经就绪，也有办法让其从select()方法返回。
            //只要让其它线程在第一个线程调用select()方法的那个对象上调用Selector.wakeup()方法即可。
            //阻塞在select()方法上的线程会立马返回。
            //如果有其它线程调用了wakeup()方法，但当前没有线程阻塞在select()方法上，
            //下个调用select()方法的线程会立即“醒来（wake up）”。
    /*
            主要作用
                解除阻塞在Selector.select()/select(long)上的线程，立即返回。
                两次成功的select之间多次调用wakeup等价于一次调用。
                如果当前没有阻塞在select上，则本次wakeup调用将作用于下一次select——“记忆”作用。
            为什么要唤醒？
                注册了新的channel或者事件。
                channel关闭，取消注册。
                优先级更高的事件触发（如定时器事件），希望及时处理。
            原理
                Linux上利用pipe调用创建一个管道，Windows上则是一个loopback的tcp连接。这是因为win32的管道无法加入select的fd set，将管道或者TCP连接加入select fd set。
                wakeup往管道或者连接写入一个字节，阻塞的select因为有I/O事件就绪，立即返回。可见，wakeup的调用开销不可忽视。
            */
            sel.wakeup();
        }
    
        @Override
        public void run() {
        
            try {
        
                if (state == READING) {
        
                    System.out.println("Reading : ");
                    read();
                } else if (state == SENDING) {
        
                    System.out.println("Sending : ");
                    send();
                }
            } catch (IOException ex) {
        
                ex.printStackTrace();
                sk.cancel();
                try {
        
                    socket.finishConnect();
                } catch (IOException e) {
        
                    e.printStackTrace();
                }
            }
    
        }
    
        void read() throws IOException {
        
            //从通道读
            int length = 0;
            while ((length = socket.read(buffer)) > 0) {
        
                System.out.println((new String(buffer.array(), 0, length)));
            }
            //读完后，准备开始写入通道,byteBuffer切换成读模式
            buffer.flip();
            //读完后，注册write就绪事件
            sk.interestOps(SelectionKey.OP_WRITE);
            //读完后,进入发送的状态
            state = SENDING;
        }
    
        void send() throws IOException {
        
            //写入通道
            socket.write(buffer);
            //写完后,准备开始从通道读,byteBuffer切换成写模式
            buffer.clear();
            //写完后,注册read就绪事件
            sk.interestOps(SelectionKey.OP_READ);
            //写完后,进入接收的状态
            state = READING;
        }
    }
    
    //上面 的实现用Handler来同时处理Read和Write事件, 所以里面出现状态判断
    //我们可以用State-Object pattern来更优雅的实现
    /*
    class Handler { // ...
        public void run() { // initial state is reader
            socket.read(input);
            if (inputIsComplete()) {
                process();
                //状态迁移, Read后变成write, 用Sender作为新的callback对象
                sk.attach(new Sender());
                sk.interestOps(SelectionKey.OP_WRITE);
                sk.selector().wakeup();
            }
        }
        class Sender implements Runnable {
            public void run(){ // ...
                socket.write(output);
                if (outputIsComplete()) sk.cancel();
            }
        }
    }
    */

*client*

### 多线程 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 3]  
多线程优化反应堆

*  handler引入线程池
 *  引入多个selector选择器，提升选择大量通道的能力。每个子subReactor子线程负责一个选择器。避免一个selector多个线程，导致进行线程同步降低效率。

class Handler implements Runnable {
        
        // uses util.concurrent thread pool
        static PooledExecutor pool = new PooledExecutor(...);
        static final int PROCESSING = 3;
        // ...
        synchronized void read() {
         // ...
            socket.read(input);
            if (inputIsComplete()) {
        
                state = PROCESSING;
                pool.execute(new Processer()); //使用线程pool异步执行
            }
        }
        
        synchronized void processAndHandOff() {
        
            process();
            state = SENDING; // or rebind attachment
            sk.interest(SelectionKey.OP_WRITE); //process完,开始等待write事件
        }
        
        class Processer implements Runnable {
        
            public void run() {
         processAndHandOff(); }
        }
    }

### 主从模式 ###

main更加专注于处理建立连接,而sub主要处理请求  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 4]

Selector[] selectors; //subReactors集合, 一个selector代表一个subReactor
    int next = 0;
    class Acceptor {
         // ...
        public synchronized void run() {
         ...
            Socket connection = serverSocket.accept(); //主selector负责accept
            if (connection != null)
                new Handler(selectors[next], connection); //选个subReactor去负责接收到的connection
            if (++next == selectors.length) next = 0;
        }
    }

#### 下面附赠我写的可跑的案例 ####

亲测有效，即拿即用。无需代码改动。(ps:最近无聊，遂手写netty，顺便构建手写的MQ模块的通信。)

package netty.reactor.master_slave;
    
    import netty.reactor.Config;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    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.Objects;
    import java.util.Set;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * 主从Reactor线程模型
     *
     * @author WangChao
     * @create 2021/5/16 23:46
     */
    public class MasterSlaveReactor implements  Runnable{
        
    
        Selector[] selectors = new Selector[2];
        SubReactor[] subReactors = new SubReactor[2];
        Selector masterSelector;
        AtomicInteger next = new AtomicInteger(0);
        ServerSocketChannel serverSocketChannel;
    
    
        public MasterSlaveReactor() throws IOException {
        
            //一个反应堆对应一个子选择器
            selectors[0] = Selector.open();
            selectors[1] = Selector.open();
            subReactors[0] = new SubReactor(selectors[0]);
            subReactors[1] = new SubReactor(selectors[1]);
    
            serverSocketChannel = ServerSocketChannel.open();
            InetSocketAddress inetSocketAddress = new InetSocketAddress(Config.HOST, Config.PORT);
            serverSocketChannel.socket().bind(inetSocketAddress);
            serverSocketChannel.configureBlocking(false);
            //第一个选择器监控accept
            masterSelector = Selector.open();
            SelectionKey sk = serverSocketChannel.register(masterSelector, SelectionKey.OP_ACCEPT);
            sk.attach(new AcceptorHandler());
    
        }
    
        private void startService() {
        
            new Thread(subReactors[0]).start();
            new Thread(subReactors[1]).start();
            run();
        }
    
        @Override
        public void run() {
        
            try {
        
                while (!Thread.interrupted()) {
        
                    masterSelector.select();
                    Set<SelectionKey> selectionKeys = masterSelector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
        
                        dispatch((SelectionKey) (iterator.next()));
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
        
            }
        }
    
        private void dispatch(SelectionKey sk) {
        
            Runnable handler = (Runnable) sk.attachment();
            if (!Objects.isNull(handler)) {
        
                handler.run();
            }
    
        }
    
        class AcceptorHandler implements Runnable {
        
            @Override
            public void run() {
        
                try {
        
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    if (!Objects.isNull(socketChannel)) {
        
                        //接收到请求创建新线程后，让子选择器进行监听处理。
                        new MasterSlaveReactorHandler(selectors[next.get()], socketChannel);
                    }
                } catch (IOException e) {
        
                }
                if (next.incrementAndGet() == selectors.length) {
        
                    next.set(0);
                }
            }
        }
    
        class SubReactor implements Runnable {
        
            private final Selector selector;
    
            public SubReactor(Selector selector) {
        
                this.selector = selector;
            }
    
            @Override
            public void run() {
        
                try {
        
                    while (!Thread.interrupted()) {
        
                        selector.select();
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while (iterator.hasNext()) {
        
                            dispatch((SelectionKey) (iterator.next()));
                        }
                        selectionKeys.clear();
                    }
                } catch (IOException e) {
        
                }
            }
    
    
        }
    
        public static void main(String[] args) throws IOException {
        
            new MasterSlaveReactor().startService();
        }
    }

Handler

package netty.reactor.master_slave;
    
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 主从Reactor线程模型，handler
     *
     * @author WangChao
     * @create 2021/5/17 0:49
     */
    public class MasterSlaveReactorHandler implements Runnable {
        
        private final SocketChannel socketChannel;
        private final SelectionKey sk;
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        static final int READING = 0, SENDING = 1, PROCESSING = 3;
        int state = READING;
    
        static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                10,
                50,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    
        public MasterSlaveReactorHandler(Selector selector, SocketChannel socketChannel) throws IOException {
        
            this.socketChannel = socketChannel;
            this.socketChannel.configureBlocking(false);
            this.sk = this.socketChannel.register(selector, 0);
            this.sk.attach(this);
            this.sk.interestOps(SelectionKey.OP_READ);
            selector.wakeup();
        }
    
        /**
         * 引入多线程后，会造成一个问题。就是重复处理多次。
         * 这是因为当监听写事件时，因为是异步发送。所以造成会重复监听多次写事件。
         */
        @Override
        public void run() {
        
            threadPoolExecutor.execute(new AsyncTask());
        }
    
        /**
         *异步线程处理，这里可能存在临界状态。
         */
        private synchronized void asyncRun() {
        
            try {
        
                if (state == READING) {
        
                    System.out.println("Reading : ");
                    int length = 0;
                    while ((length = socketChannel.read(buffer)) > 0) {
        
                        System.out.println(new String(buffer.array(), 0, length));
                    }
                    buffer.flip();
                    sk.interestOps(SelectionKey.OP_WRITE);
                    state = SENDING;
                } else if (state == SENDING) {
        
                    System.out.println("Sending : ");
                    socketChannel.write(buffer);
                    buffer.clear();
                    sk.interestOps(SelectionKey.OP_READ);
                    state = READING;
                }
                //取消监听的事件,这里注释是为了重复使用
    //            sk.cancel();
            } catch (IOException e) {
        
                e.printStackTrace();
                sk.cancel();
                try {
        
                    socketChannel.finishConnect();
                } catch (IOException ex) {
        
                    ex.printStackTrace();
                }
            }
        }
    
        class AsyncTask implements Runnable {
        
            @Override
            public void run() {
        
                MasterSlaveReactorHandler.this.asyncRun();
            }
        }
    
    
    }

## Proactor模型 ##

Proactor模型，需要先注册I/O完成事件，同时申请一片用户空间用于存储待接收的数据。调用读操作，当网卡接收到数据时，DMA将数据从网卡缓冲区直接传输到用户缓冲区，然后产生完成通知，读操作即完成。

异步I/O采用直接输入I/O或直接输出I/O，用户缓存地址会传递给设备驱动程序，数据会直接从用户缓冲区读取或直接写入用户缓冲区，相比缓冲I/O减少内存复制。

[C_select_poll_epoll_l]: https://blog.csdn.net/Nuan_Feng/article/details/108877713
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70]: /images/20230209/c5c45537320b47aa91d16a92a49446d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 1]: /images/20230209/b08d2ba0f88c430b9b60fb46f80cf115.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center]: /images/20230209/57cd5d054c82497ca0ea35126805f0dc.png
[Doug Lea]: http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20230209/e6b5750fc5044113bbb1d33237830c07.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20230209/963ebd81aabd4a118f36fe4020c40b2c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20230209/fa80908be73f4883924cf4f115349b47.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20230209/23ba1884c275451db1817da502246430.png