• JavaNIO——多线程以及IO模型（笔记）

    文章目录

    一、多线程优化

    1.1 Boss与Worker

    Boss：负责建立连接

    • thread
    • selector

    Worker：worker负责数据的读写

    • thread
    • selector

    在编写多线程时，我们要注意一个执行顺序

    1. selector.open
    2. socketChannel.register
    3. thread.start，包含selector.select
    4. selector.select

    1.2 初步实现1Boss，1Worker

    Client

    package com.yjx23332.netty.test;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class TestClient {
    public static void main(String[] args) throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress("localhost",8080));
        socketChannel.write(Charset.defaultCharset().encode("helloword"));
        System.in.read();
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    Boss

    package com.yjx23332.netty.test;



import lombok.extern.slf4j.Slf4j;

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;


/**
 * Boss角色
 * */
@Slf4j
public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey bossKey = serverSocketChannel.register(boss, SelectionKey.OP_ACCEPT, null);
        serverSocketChannel.bind(new InetSocketAddress(8080));
        //创建worker
        Worker worker = new Worker("worker-0");
        while (true) {
            boss.select();
            Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                if (key.isAcceptable()) {
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    log.debug("connected...{}",socketChannel.getRemoteAddress());
                    //2. 关联 selector
                    log.debug("before register..{}",socketChannel.getRemoteAddress());
                    worker.register(socketChannel,SelectionKey.OP_READ,null);
                    log.debug("after register..{}",socketChannel.getRemoteAddress());
                }
            }
        }
    }
}



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50

    worker

    package com.yjx23332.netty.test;


import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

import static com.yjx23332.netty.test.ByteBufferUtil.debugAll;

@Slf4j
@Data
class Worker implements Runnable{
    private Selector selector;
    volatile private Thread thread;
    private String name;
    public Worker(String name){
        this.name = name;
    }

    //把同步队列作为消息队列
    private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

    //初始化线程和selector
    public void register(SocketChannel socketChannel,int selectionKey,Object attachment) throws IOException {
    //懒汉模式
        if(this.thread == null) {
            synchronized (this) {
                if (this.thread == null) {
                    this.thread = new Thread(this, this.name);
                    this.selector = Selector.open();
                    this.thread.start();
                }
            }
        }
        //加入消息队列
        queue.add(()->{
            try {
                socketChannel.register(selector,selectionKey,attachment);
            } catch (ClosedChannelException e) {
                throw new RuntimeException(e);
            }
        });
        //唤醒select
        selector.wakeup();
    }

    @Override
    public void run() {
        SelectionKey selectionKey = null;
        while(true){
            try {
                selector.select();
                //从消息队列中取出
                Runnable task = queue.poll();
                if(task != null){
                    task.run();//执行注册
                }
                Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                while(iter.hasNext()){
                    selectionKey = iter.next();
                    iter.remove();
                    if(selectionKey.isReadable()){
                        ByteBuffer buffer = ByteBuffer.allocate(16);
                        SocketChannel channel = (SocketChannel) selectionKey.channel();
                        log.debug("read...{}",channel.getRemoteAddress());
                        channel.read(buffer);
                        buffer.flip();
                        debugAll(buffer);
                    }
                }
            }
            catch (IOException e) {
                log.debug("{}",e.getStackTrace());
                if(selectionKey != null)
                    selectionKey.cancel();
            }
            catch (Exception e){
                throw new RuntimeException(e);
            }
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90

    1.3 线程数目配置

    package com.yjx23332.netty.test;



import lombok.extern.slf4j.Slf4j;

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.concurrent.atomic.AtomicInteger;


/**
 * Boss角色
 * */
@Slf4j
public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey bossKey = serverSocketChannel.register(boss, SelectionKey.OP_ACCEPT, null);
        serverSocketChannel.bind(new InetSocketAddress(8080));


        //创建特定数量的Worker,我们用CPU核心数目来设置线程数
        //切记：在docker容器中，因为docker不是物理隔离的，因此会获取物理CPU个数，而不是容器申请的个数
        //jdk10才解决这个问题，用JVM参数UserContainerSupport配置，默认开启
        Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];
        for(int i = 0;i < workers.length;i++){
            workers[i] = new Worker("worker-" + i );
        }

        AtomicInteger atomicIntegers = new AtomicInteger();

        while (true) {
            boss.select();
            Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                if (key.isAcceptable()) {
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);

                    int index = atomicIntegers.getAndUpdate(e->{
                        return (e + 1) % workers.length;
                    });
                    //round robin 轮询负载均衡
                    workers[index].register(socketChannel,SelectionKey.OP_READ,null);
                    log.info("当前使用的是：" + index);
                }
            }
        }
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62

    二、UDP

    • UDP是无连接的，client发送数据不会管Server是否开启
    • server这边的receive方法会将收到的数据存入bytebuffer，但如果数据报文超过buffer大小，多出来的数据则会被丢弃。

    因为他没有建立可靠连接，所以我们传输与之前的TCP模式不一样。

    package com.yjx23332.netty.test;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.charset.Charset;
@Slf4j
public class TestClient {
    public static void main(String[] args) throws IOException {
        DatagramChannel datagramChannel = DatagramChannel.open();
        datagramChannel.connect(new InetSocketAddress("localhost",9999));
        datagramChannel.write(Charset.defaultCharset().encode("hello,word!"));
        ByteBuffer byteBuffer = ByteBuffer.allocate(200);
        datagramChannel.read(byteBuffer);
        byteBuffer.flip();
        log.debug("收到来自服务器消息：{}",Charset.defaultCharset().decode(byteBuffer));
        byteBuffer.clear();
        datagramChannel.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    package com.yjx23332.netty.test;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.charset.Charset;
import java.util.Iterator;


@Slf4j
public class UDPServer {
    public static void main(String[] args){
        try(DatagramChannel datagramChannel = DatagramChannel.open();){
            datagramChannel.bind(new InetSocketAddress(9999));
            datagramChannel.configureBlocking(false);
            ByteBuffer byteBuffer = ByteBuffer.allocate(200);
            Selector selector = Selector.open();
            datagramChannel.register(selector, SelectionKey.OP_READ);
            while (true) {
                selector.select();
                Iterator<SelectionKey> it = selector.keys().iterator();
                while (it.hasNext()) {
                    SelectionKey selectionKey = it.next();
                    if(selectionKey.isReadable()){
                        DatagramChannel channel = (DatagramChannel) selectionKey.channel();
                        //receive 会返回socket地址
                        InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.receive(byteBuffer);
                        byteBuffer.flip();
                        log.debug("The message received is ：{}", Charset.defaultCharset().decode(byteBuffer));
                        byteBuffer.clear();
                        // 通过socket返回
                        datagramChannel.send(Charset.defaultCharset().encode("Server has received the message!"), inetSocketAddress);
                    }
                    //移除该事件
                    selector.selectedKeys().remove(selectionKey);

                }
            }
        }catch(IOException ioException){
            ioException.printStackTrace();
        }
    }

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50

    三、IO模型

    3.1 stream（BIO） vs channel（NIO）

    • Stream不会自动缓冲数据，channel会利用系统提供的发送缓冲区、接收缓冲区（更为底层）
    • Stream仅支持阻塞API。channel同时支持阻塞、非阻塞API，网络channel可配合selector实现多路复用
    • 二者均为全双工

    3.2 IO模型

    当调用一次channel.read或者stream.read后，会切换至操作熊内核态完成真正的树读取，而读取又分为

    • 等待数据阶段

    • 复制数据阶段

    3.2.1 阻塞IO

    1. 用户线程发起read（用户线程）
    2. 阻塞等待数据（内核空间，阻塞用户线程）
    3. 从内核复制数据给用户（内核空间，阻塞用户线程）
    4. 用户线程继续执行（用户线程）

    3.2.2 非阻塞IO

    1. 用户线程发起read（用户线程）
    2. 有数据就从内核复制数据给用户，无数据就返回约定的值（内核空间，阻塞用户线程）
    3. 用户线程继续执行（用户线程）

    因此我们通常会在加一个循环反复执行上面的逻辑。

    3.3.3 多路复用

    1. 用户线程select(用户线程发起事件监听。无参则是等待内核唤醒，有参则是等待一定时间后自动唤醒）
    2. 内核阻塞select（内核，阻塞用户线程）
    3. 内核告知有事件发生或者到时间了（内核）
    4. 用户线程依据key调用read（用户线程）
    5. 从内核复制数据给用户（内核空间，阻塞用户线程）
    6. 用户线程继续执行（用户线程）

    多路复用比之阻塞IO优势？

    1. 阻塞IO是针对读写、建立连接这一种单一行为。先建立连接、在读取数据这样进行循环。此时我们已建立连接的消息到达将被阻塞，直到有一个建立连接，他才会被一起处理。
    2. 多路复用是针对整个流程的阻塞。它监视多个channel的事件。无论哪一个事件都会触发执行。他比起阻塞IO减少了无意义的循环对CPU的占用，又比阻塞IO只针对某种事件的阻塞有效。

    3.3.4 信号驱动

    1. 用户线程预先向内核设置信号（用户线程）
    2. 内核设置好了相关信息后返回参数，让用户线程知道设置结果（内核）
    3. 等待数据（内核等待，用户线程继续执行）
    4. 收到数据后，发送信号给用户线程（内核）
    5. 用户线程调用回调方法处理（用户线程）
    6. 从内核复制数据给用户（内核，阻塞用户线程）
    7. 用户线程接收结果（用户线程）

    它前半段异步，后半段阻塞。

    • 优势：进程没有收到SIGIO信号之前，不被阻塞，可以做其他事情。
    • 劣势：数据量变大时，信号产生太频繁，性能较低。

    3.3.5 异步IO（AIO）

    • 同步：线程自己去获取结果（一个线程)。
    • 异步：线程自己不去获取结果，而是其他线程送来结果（至少连两个线程）。
    1. 用户线程通过用一个线程发起read（用户线程）
    2. 通过回掉方法返回参数，用来之后获取结果。等待数据（内核空间处理，而用户线程继续执行）
    3. 从内核复制数据给用户（内核空间处理，而用户线程继续执行）
    4. 通过一个线程的回调方法返回真正结果（内核）
    • 同步阻塞：阻塞IO
    • 同步非阻塞：非阻塞IO
    • 异步非阻塞：AIO
    • 异步阻塞：多路复用（这里其实分的不是很清楚，使用的是另一个线程去接收消息，而自己在这边等着。但同时，它又使用的非阻塞IO处理事件。）
    • 信号驱动（使用的是另一个线程去接收消息，自己去做其他事情）

    3.3 零拷贝

    3.3.1 传统IO问题

    将一个文件通过socket写出

    	public static void main(String[] args) throws IOException {
        RandomAccessFile randomAccessFile = new RandomAccessFile("data.txt","r");
        byte[] buf = new byte[(int)randomAccessFile.length()];
        randomAccessFile.read(buf);
        Socket socket = new Socket("localhost",8080);
        socket.getOutputStream().write(buf);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    工作流程

    1
    2
    3
    4
    磁盘
    内核缓冲区
    用户缓冲区
    socket缓冲区
    网卡
    1. Java本身不具备IO读写能力，因此read方法调用后，要从Java程序的用户态切换到内核态，去调用操作系统（Kernel）的读能力，将数据读入内核缓冲区。这期间用户线程阻塞，操作系统使用MDA（Direct Memory Access）来实现文件读，期间不会调用CPU。

    DMA硬件单元。计算机中有一些硬件转么处理一些简单的读写，USB连接等，以减少CPU的占用，只需要最后告知CPU相关结果。

    ２.内核态切换回用户态，将数据从内核缓冲区读入用户缓冲区（byte[] buf），这期间CPU会参与拷贝，无法利用DMA。

    ３.调用write方法，这时将数据从用户缓冲区（byte[]）写入socket缓冲区，CPU会参与拷贝

    4.接下来会访问网卡写数据，此时Java又不具备该能力，因此将从用户态转换至内核态，调用操作系统的写能力，使用DMA将socket缓冲区的数据写入网卡，该过程也不会使用CPU。

    • 3次状态切换
    • 4次拷贝

    3.3.2 NIO的处理方式

    通过DirectByteBuf

    • ByteBuffer.allocate(10).HeapByteBuffer(JVM堆内存）
    • ByteBuffer.allocate(10).DirectByteBuffer（操作系统内存，操作系统也可以访问）

    工作流程

    1
    2
    3
    磁盘
    内核缓冲区/用户缓冲区
    socket缓冲区
    网卡

    这里DirectByteBuffer将作为内核缓冲区与用户缓冲区的公用区域。但是仍然需要切换回用户态创建。

    • 3次拷贝
    • 3次状态切换

    3.3.3 Linux优化

    linux提供sendFile方法，可以将文件直接发向目标区域。

    也即是FileChannel中的transferTo/transferFrom方法底层调用的方法，我们不需要再调用Java创建一个缓存来存储了。

    工作流程

    1
    2
    3
    磁盘
    内核缓冲区
    socket缓冲区
    网卡
    1. Java调用transferTo/transferFrom方法后，切换为内核态，用DMA将数据读入内核缓冲区
    2. 数据从内核缓冲区写入socket缓冲区
    3. 数据从socket缓冲区写入网卡
    • 1次状态切换
    • 3次数据拷贝

    3.3.4 Linux（2.4）的再次优化

    1
    2
    2
    磁盘
    内核缓冲区
    socket缓冲区
    网卡
    1. 调用transferTo/transferFrom方法后，从用户态切换为内核态，用DMA将数据读入内核缓冲区
    2. 将一些偏移量、长度信息考入socket缓冲区，几乎无消耗
    3. 使用DMA将内核缓冲区的数据写入网卡，不会使用CPU
    • 1次状态切换
    • 2次数据拷贝

    3.3.3与3.3.4都可以称作零拷贝。这里零指的是Java中进行的拷贝次数。
    优点：

    1. 更少的状态切换
    2. 不利用CPU计算，减少CPU缓存伪共享
    3. 零拷贝适合小文件传输

    • CPU缓存伪共享:

      • Cache缓存很珍贵，一般分为多个缓存行。以缓存行（cache line）为单位进行管理。
      • 1缓存行中可能存放多个数据，它们可以来自于多个线程。当一个线程更新自己的内容时，对应的缓存行就会检测到变化，更新后提示相关线程。
      • 这就导致了，一个线程的数据即使没有更新，但是在相同缓存行，也得更新一下。特别是我们用了volatile的时候。
      • 我们要避免，可以额外为一个线程创建数据，把一个缓存行占满。或者用@Content注解

    3.4 AIO

    AIO用来解决数据复制阶段的阻塞问题。

    • 同步意味着，在进行读写操作时，线程需要等待结果，还是相当于闲置。
    • 异步意味着，在进行读写操作时，线程不必等待结果，而是将来又操作系统来通过回调方式又另外的线程来获的结果。

    异步模型需要底层操作系统（Kernel）提供支持
    windows系统通过IOCP实现了真正的异步IO
    LInux系统异步IO在2.6版本引入，但其底层实现还是用多路复用模拟了异步IO，性能没有优势

    可以参考IO模型之AIO代码及其实践详解

    3.4.1 文件AIO

    package com.yjx23332.netty.test;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;


import static com.yjx23332.netty.test.ByteBufferUtil.debugAll;


@Slf4j
public class AioDemo {
    public static void main(String[] args){
        try(AsynchronousFileChannel asynchronousFileChannel = AsynchronousFileChannel.open(Paths.get("data.txt"), StandardOpenOption.READ)){
            /**
             * @param ByteBuffer,start,attachment,callbackfunction
             *
             */
            ByteBuffer buffer = ByteBuffer.allocate(16);
            asynchronousFileChannel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                //一次read操作成功，就调用一次
                /**
                 * result读取的实际字节数，attachment
                 * */
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    log.debug("开始读取...");
                    attachment.flip();
                    debugAll(attachment);
                }
                //异常则调用
                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                        log.info("{}",exc);
                }
            });
            log.debug("执行到最后了...");
            System.in.read();
        }catch (IOException ioException){
            log.error("{}",ioException.getStackTrace());
        }
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49

    参考目录

    [1]黑马程序员Netty全套教程

  • 相关阅读:
    指针进阶全
    第02章 BeautifulSoup 入门
    el-select 多选模式下嵌套el-tree 删除tag时能去掉el-tree对应节点的勾
    全志XR806+TinyMaix，在全志XR806上实现ML推理
    【无标题】
    武汉链（基于ETH）BSN官方DDC链上数据解析
    Java NIO与IO的区别和比较
    请你详细说说类加载流程，类加载机制及自定义类加载器
    判断是否为2的幂
    Codeforces Round #788 (Div. 2)
  • 原文地址：https://blog.csdn.net/weixin_46949627/article/details/126664651