BIO&NIO&Netty初步个人笔记

参考 [b站尚硅谷韩顺平老师视频,文末获取原生笔记]

BIO编程简单流程

1. 服务器端启动一个ServerSocket
2. 客户端启动Socket对服务器进行通 信，默认情况下服务器端需要对每 个客户 建立一个线程与之通讯
3. 客户端发出请求后, 先咨询服务器 是否有线程响应，如果没有则会等 待，或者被拒绝
4. 如果有响应，客户端线程会等待请 求结束后，在继续执行

package BIOprogram;

import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ServerBIO {
    public static void main(String[] args) throws IOException {
        // 线程池机制
        // 创建线程池 客户端连接就创建一个线程与之通讯
        ExecutorService executorThreadPool = Executors.newCachedThreadPool();
        // 创建serverSocket
        ServerSocket serverSocket = new ServerSocket(6666);
        System.out.println("服务端启动");
        // 无限循环，创建新线程
        while (true) {
            System.out.println("线程信息 id:" + Thread.currentThread().getId() + "线程名字:" + Thread.currentThread().getName());
            // 监听连接
            System.out.println("等待连接.......");
            final Socket accept = serverSocket.accept();
            System.out.println("连接到一个新客户");
            // 创建一个线程与之通信
            executorThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    // 和客户端通信方法
                    handler(accept);
                }
            });
        }
    }

    public static void handler(Socket socket) {
        try {
            System.out.println("线程id:" + Thread.currentThread().getId() + "--线程名字:" + Thread.currentThread().getName());
            byte[] bytes = new byte[1024];
            InputStream inputStream = socket.getInputStream();
            while (true) {
                System.out.println("线程id:" + Thread.currentThread().getId() + "--线程名字:" + Thread.currentThread().getName());
                System.out.println("读取客户端数据......");
                int len = inputStream.read(bytes);
                if (len != -1) {
                    System.out.println(new String(bytes,0,len));
                }else {
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭连接
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}

tg:客户端用telnet模拟

NIO 非阻塞 网络编程原理分析图

NIO 有三大核心部分：Channel(通道)，Buffer(缓冲区), Selector(选择器)
三者关系

通道(Channel)
应用实例1-本地文件写数据 实例要求:

1. 使用前面学习后的ByteBuffer(缓冲) 和 FileChannel(通道)， 将 “hello,尚硅谷” 写入 到file01.txt 中
2. 文件不存在就创建

package channelTest;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;


public class Anli01 {
    public static void main(String[] args) throws IOException {
        String str = "hello,尚硅谷";
        // 文件输出流写数据
        FileOutputStream outputStream = new FileOutputStream("file01.txt",true);
        // 获取filechannel
        FileChannel channel = outputStream.getChannel();
        // 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        // 将数据写入缓冲区
        buffer.put(str.getBytes(StandardCharsets.UTF_8));
        // buffer 进行 flip
        buffer.flip();
        // 将buffer 中数据写入到filechannel
        channel.write(buffer);
        // 关闭流
        outputStream.close();
    }
}

应用实例2-本地文件读数据 实例要求:

1. 使用前面学习后的ByteBuffer(缓冲) 和 FileChannel(通道)， 将 file01.txt 中的数据读 入到程序，并显示在控制台屏幕
2. 假定文件已经存在

package channelTest;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;


public class Anli02 {
    public static void main(String[] args) throws IOException {
        File file = new File("G:\netty\netty-proj\file01.txt");
        FileInputStream fileInputStream = new FileInputStream(file);
        FileChannel channel = fileInputStream.getChannel();
        ByteBuffer buffer = ByteBuffer.allocate((int)file.length());
        // 将channel里数据读到buffer
        channel.read(buffer);
        // 将buffer里二进制数据转为字符串
        String s = new String(buffer.array());
        System.out.println(s);
        fileInputStream.close();
    }
}

应用实例3-使用一个Buffer完成文件读取 实例要求:

1. 使用 FileChannel(通道) 和 方法 read , write，完成文件的拷贝
2. 拷贝一个文本文件 1.txt , 放在项目下即可

package channelTest;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;


public class Anli03 {
    public static void main(String[] args) throws IOException {
        FileInputStream fileInputStream = new FileInputStream("file01.txt");
        FileChannel channel01 = fileInputStream.getChannel();
        FileOutputStream fileOutputStream = new FileOutputStream("file02.txt");
        FileChannel channel02 = fileOutputStream.getChannel();
        // 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        // 读写数据操作 读一次写一次
        while (true) { // 无限循环读数据
            
            // clear 重置
            buffer.clear();
            // 读到buffer read为读到数据大小 这里为文件大小
            int read = channel01.read(buffer);
            System.out.println("read = " + read);
            if (read == -1) {
                break;
            }
            // 写
            buffer.flip();
            channel02.write(buffer);
        }
        fileInputStream.close();
        fileOutputStream.close();
    }
}

应用实例4-拷贝文件transferFrom 方法 实例要求:

1. 使用 FileChannel(通道) 和 方法 transferFrom ，完成文件的拷贝
2. 拷贝一张图片

package channelTest;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.channels.FileChannel;


public class Anli04 {
    public static void main(String[] args) throws IOException {
        FileInputStream fileInputStream = new FileInputStream("cat.jpeg");
        FileOutputStream fileOutputStream = new FileOutputStream("copy-cat.jpeg");
        // 创建channel
        FileChannel source = fileInputStream.getChannel();
        FileChannel des = fileOutputStream.getChannel();
        // 通道transForm方法 拷贝
        des.transferFrom(source, 0, source.size());
        // 关闭通道和流
        des.close();
        source.close();
        fileInputStream.close();
        fileOutputStream.close();
    }
}

使用多个buffer，buffer数组
Scattering和Gathering 分散和聚合

package channelTest;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.Date;


public class ScatteringAndGathering {
    public static void main(String[] args) throws IOException {
        // 创建serversocketchaael网络
        ServerSocketChannel socketChannel = ServerSocketChannel.open();
        InetSocketAddress inetSocketAddress = new InetSocketAddress(7777);
        // 绑定端口并启动
        socketChannel.socket().bind(inetSocketAddress);
        // 创建buffer数组
        ByteBuffer[] byteBuffers = new ByteBuffer[2];
        byteBuffers[0] = ByteBuffer.allocate(3);  //设置每个buffer大小
        byteBuffers[1] = ByteBuffer.allocate(5);
        // 连接客户端 telnet
        System.out.println("等待连接.....");
        SocketChannel accept = socketChannel.accept();
        System.out.println(new Date()+"新连接:"+accept.getLocalAddress());
        // 设置读取数据大小
        long readMes = 8;
        // while循环读取
        while (true) {
            long readByte = 0;
            while (readByte < readMes) {
                long read = accept.read(byteBuffers);
                readByte += read;
                System.out.println("readByte=" + readByte);
                // 使用流打印,查看position，limit
                Arrays.asList(byteBuffers).stream().map(buffer -> "position:" + buffer.position() + "limit:" + buffer.limit()).forEach(System.out::println);
            }
            // 客户端
            Arrays.asList(byteBuffers).forEach(buffer -> buffer.flip()); // 将每个buffer flip
            long writeLen = 0;
            while (writeLen < readMes) {
                long write = accept.write(byteBuffers);
                writeLen += write;
            }
            // 将buffer进行clear重置
            Arrays.asList(byteBuffers).forEach(buffer->buffer.clear());
            System.out.println("writeLen:"+writeLen+"readByte"+readByte);
        }
    }
}

selector 相关方法说明
selector.select()//阻塞
selector.select(1000);//阻塞1000毫秒，在1000毫秒后返回 selector.wakeup();//唤醒
selector selector.selectNow();//不阻塞，立马返还
SelectionKey，表示 Selector 和网络通道的注册关系, 共四种:
int OP_ACCEPT：有新的网络连接可以 accept，值为 16
int OP_CONNECT：代表连接已经建立，值为 8
int OP_READ：代表读操作，值为 1
int OP_WRITE：代表写操作，值为 4
案例要求:

1. 编写一个 NIO 入门案例，实现服务器端和客户端之间的数据简单通讯（非阻塞）
2. 目的：理解NIO非阻塞网络编程机制

服务端代码逻辑

package NIOoperation;

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.Date;
import java.util.Iterator;
import java.util.Set;


public class NIOServer {
    public static void main(String[] args) throws IOException {
        // 创建并启动serversocketchannel
        ServerSocketChannel socketChannel = ServerSocketChannel.open();
        // 创建selector
        Selector selector = Selector.open();
        // 绑定channel端口
        socketChannel.socket().bind(new InetSocketAddress(7777));
        // 设置为非阻塞
        socketChannel.configureBlocking(false);
        // 将channel注册到selector accept
        socketChannel.register(selector, SelectionKey.OP_ACCEPT);
        // while循环 等待连接
        while (true) {
            // 阻塞一秒 等待连接
            if (selector.select(1000) == 0) {
                System.out.println("服务器等待一秒，无连接");
                continue;
            }
            // 获取selectionKeys 集合
            Set selectionKeys = selector.selectedKeys();
            // 使用迭代器迭代
            Iterator iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                // 判断为accept事件
                SelectionKey next = iterator.next();
                if (next.isAcceptable()) {
                    // 客户端有新连接
                    SocketChannel accept = socketChannel.accept();
                    System.out.println(new Date() + ":客户端方面新建一个连接[" + accept.getLocalAddress() + "]");
                    // 设置为非阻塞
                    accept.configureBlocking(false);
                    // 注册read 事件到selector 并关联一个buffer
                    accept.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                }
                // read事件
                if (next.isReadable()) {
                    // 从selector集合获取read
                    SocketChannel channel = (SocketChannel) next.channel();
                    // 从channel中获取关联buffer
                    ByteBuffer buffer = (ByteBuffer) next.attachment();
                    // 从channel中读数据
                    channel.read(buffer);
                    System.out.println("客户端来的数据："+new String(buffer.array()));
                }
                // 手动移除selector
                iterator.remove();
            }

        }
    }

}

客户端逻辑

package NIOoperation;

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

public class NIOClient {
    public static void main(String[] args) throws IOException {
        // 创建并启动socketchannel
        SocketChannel socketChannel = SocketChannel.open();
        // 设置非阻塞
        socketChannel.configureBlocking(false);
        // 设置服务器端 ip 和 端口
        InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 7777);
        // 非阻塞可视代码 连接服务端
        if (!socketChannel.connect(inetSocketAddress)) {
            while (!socketChannel.finishConnect()) {
                System.out.println("非阻塞方式，可以做其他事");
            }
        }
        // 连接成功
        String str = "hello,world";
        ByteBuffer buffer = ByteBuffer.wrap(str.getBytes(StandardCharsets.UTF_8));
        socketChannel.write(buffer);
        System.in.read();
    }
}

NIO 网络编程应用实例-群聊系统 实例要求:

1. 编写一个 NIO 群聊系统，实现服务器 端和客户端之间的数据简单通讯（非 阻塞）
2. 实现多人群聊
3. 服务器端：可以监测用户上线，离线， 并实现消息转发功能
4. 客户端：通过channel 可以无阻塞发送 消息给其它所有用户，同时可以接受 其它用户发送的消息(有服务器转发得 到)
5. 目的：进一步理解NIO非阻塞网络编程 机制
   

服务端

package NIOoperation;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;


// 服务端
public class GroupChatServer {
    // 定义属性 selector channel port
    private Selector selector;
    private ServerSocketChannel serverSocketChannel;
    private static final int PORT = 7777;

    // 构造器 初始化方法 build channel(!congestion) & selector and bind(port) and set selectionKey
    public GroupChatServer() {
        try {
            // 创建selector channel
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            // 非阻塞
            serverSocketChannel.configureBlocking(false);
            // 绑定端口
            serverSocketChannel.bind(new InetSocketAddress(PORT));
            // 注册accept事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 监听
    public void listen() {
        // 循环处理
        try {
            while (true) {
                // 需要处理事件数量
                int num = selector.select();
//                System.out.println(num);
                if (num > 0) {
                    // 取出事件集合
                    Iterator iterator = selector.selectedKeys().iterator();
                    // 遍历集合
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        // 连接事件
                        if (key.isAcceptable()) {
                            // 获取连接channel
                            SocketChannel channel = serverSocketChannel.accept();
                            // 设置非阻塞
                            channel.configureBlocking(false);
                            System.out.println(new Date() + ": " + channel.getRemoteAddress() + "已上线");
                            // 注册read
                            channel.register(selector, SelectionKey.OP_READ);
                        }
                        // read事件
                        if (key.isReadable()) {
                            // 分离读取事件
                            readData(key);
                        }
                        iterator.remove();
                    }
                        // 移除key
//                        selectionKeys.remove(key);

                }else {
                    System.out.println("等待连接...");
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 异常处理
        }
    }

    public void readData(SelectionKey key) {
        SocketChannel socketChannel = null;
        try {
            // 得到客户端channel
            socketChannel = (SocketChannel)key.channel();
            // 得到数据
//        ByteBuffer buffer = (ByteBuffer)key.attachment();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int num = socketChannel.read(buffer);
            if (num > 0) {
                // 转化
                String str = new String(buffer.array());
                // 输出数据
                System.out.println(new Date()+" ip "+str.trim());  // trim
                // 向其他客户端转发消息 另定义一个方法
                sendInfoToOthers(str, socketChannel);
            }
        } catch (IOException e) {
            try {
                System.out.println(socketChannel.getRemoteAddress() + "离线了...");
                // 取消注册
                key.cancel();
                // 关掉channel
                socketChannel.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendInfoToOthers(String msg, SocketChannel channel) throws Exception{
        System.out.println("服务端向其他客户端转发消息ing...");
        // 遍历所有selector上的channel 并排除自身
        for (SelectionKey key : selector.keys()) {
            // 取出各channel
            Channel selectableChannel = key.channel();
            if (selectableChannel instanceof SocketChannel && selectableChannel != channel) {
                // 转换
                SocketChannel socketChannel = (SocketChannel) selectableChannel;
                // 将消息放到buffer
                ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes(StandardCharsets.UTF_8));
                // 写入
                socketChannel.write(buffer);
            }
        }
    }

    public static void main(String[] args) {
        // 创建服务端对象
        GroupChatServer groupChatServer = new GroupChatServer();
        groupChatServer.listen();
    }
}

客户端

package NIOoperation;

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.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;

public class GroupChatClient {
    // 定义属性 IP and PORT and selector and socketchannel and username
    private final String IP = "127.0.0.1";
    private final int PORT = 7777;
    private Selector selector;
    private SocketChannel socketChannel;
    private String username;

    // 构造器 初始化 创建selector & channel 非阻塞 连接服务端 注册channel 设置username
    public GroupChatClient() throws IOException {
        selector = Selector.open();
        // 直接连接
        socketChannel = SocketChannel.open(new InetSocketAddress(IP,PORT));
        // 非阻塞
        socketChannel.configureBlocking(false);
        // 注册
        socketChannel.register(selector, SelectionKey.OP_READ);
        // 设置username
        username = socketChannel.getLocalAddress().toString().substring(1);
        System.out.println(username + " is ok");
    }

    // 向服务器发送消息
    public void sendInfo(String msg) {
        msg = username + " : " + msg;
        try {
//            ByteBuffer buffer = ByteBuffer.allocate(1024);
//            buffer.put(msg.getBytes(StandardCharsets.UTF_8));
            socketChannel.write(ByteBuffer.wrap(msg.getBytes()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 获取服务端回复消息
    public void readInfo() {
        try {
            // 从selector中获取
            int num = selector.select();
//            System.out.println(num);
            if (num > 0) {
                // 获取selectionKey
                Iterator iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    if (key.isReadable()) {
                        // 获取channel
                        SocketChannel channel = (SocketChannel) key.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        // 读数据
                        channel.read(buffer);
                        // 转换
                        String msg = new String(buffer.array());
                        // 输出
                        System.out.println("ip【"+channel.getLocalAddress()+"】"+"发来一个消息: " + msg.trim());
                    }
                    iterator.remove();
                    // 移除key
//                    selectionKeys.remove(key);
                }
            }else {
                System.out.println("没有可用channel");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException {
        // 启动客户端
        GroupChatClient groupChatClient = new GroupChatClient();
        // 设置新线程 每隔3秒从服务器读取数据
        new Thread() {
            public void run() {
                // 循环 不断读数据
                while (true) {
                    groupChatClient.readInfo();
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }.start();
        Scanner sc = new Scanner(System.in);
        while (sc.hasNextLine()) {
            String msg = sc.nextLine();
            groupChatClient.sendInfo(msg);
        }
//        groupChatClient.sendInfo("1");
    }
}

NIO与零拷贝 零拷贝基本介绍

1. 零拷贝是网络编程的关键，很多性能优化都离不开。
2. 在 Java 程序中，常用的零拷贝有 mmap(内存映射) 和 sendFile。

mmap 优化
mmap 通过内存映射，将文件映射到内核缓冲区，同时，用户空间可以共享内核 空间的数据。这样，在进行网络传输时，就可以减少内核空间到用户控件的拷贝 次数。如下图

sendFile 优化
Linux 2.1 版本 提供了sendFile 函数，其基本原理如下：数据根本不经过用户态，直接从内核缓冲区进入到 SocketBuffer，同时，由于和用户态完全无关，就减少了一次上下文切换。

零拷贝的再次理解

1. 我们说零拷贝，是从操作系统的角度来说的。因为内核缓冲区之间，没有数据是重复的（只有 kernel buffer 有一份数据）。
2. 零拷贝不仅仅带来更少的数据复制，还能带来其他的性能优势，例如更少的上下文切换，更少的 CPU 缓存伪共享以及无 CPU 校验和计算。

mmap 和 sendFile 的区别

1. mmap 适合小数据量读写，sendFile 适合大文件传输。
2. mmap 需要 4 次上下文切换，3 次数据拷贝；sendFile 需要 3 次上下文切换，最少 2 次数据拷贝。
3. sendFile 可以利用 DMA 方式，减少 CPU 拷贝，mmap 则不能（必须从内核拷贝到 Socket 。

NIO 零拷贝案例
案例要求：

1. 使用传统的IO 方法传递一个大文件
2. 使用NIO 零拷贝方式传递(transferTo)一个大文件
3. 看看两种传递方式耗时时间分别是多少
   服务端

package NIOoperation.zerocopy;

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

public class NewIoServer {
    public static void main(String[] args) throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress( 7777));
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        while (true) {
            SocketChannel socketChannel = serverSocketChannel.accept();
            int read = 0;
            while (read != -1) {
                socketChannel.read(buffer);
                buffer.rewind();
            }
        }
    }
}

客户端

package NIOoperation.zerocopy;

import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;

public class NewIoClient {
    public static void main(String[] args) throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress("localhost", 7777));
        FileChannel channel = new FileInputStream("软件apk.rar").getChannel();
        long start = System.currentTimeMillis();
        // 零拷贝 transferto
        channel.transferTo(0, channel.size(), socketChannel);
        long end = System.currentTimeMillis();
        System.out.println("拷贝数据大小:"+channel.size()+"--用时:"+(end-start));
    }
}

Java AIO 基本介绍

1. JDK 7 引入了 Asynchronous I/O，即 AIO。在进行 I/O 编程中，常用到两种模式：Reactor和 Proactor。Java 的 NIO 就是 Reactor，当有事件触发时，服务器端得到通知，进行相应的处理。
2. AIO 即 NIO2.0，叫做异步不阻塞的 IO。AIO 引入异步通道的概念，采用了Proactor 模式，简化了程序编写，有效的请求才启动线程，它的特点是先由操作系统完成后才通知服务端程序启动线程去处理，一般适用于连接数较多且连接时间较长的应用。
3. 目前 AIO 还没有广泛应用，Netty 也是基于NIO, 而不是AIO。

原生NIO存在的问题

1. NIO 的类库和 API 繁杂，使用麻烦：需要熟练掌握 Selector、ServerSocketChannel、SocketChannel、ByteBuffer 等。
2. 需要具备其他的额外技能：要熟悉 Java 多线程编程，因为 NIO 编程涉及到 Reactor模式，你必须对多线程和网络编程非常熟悉，才能编写出高质量的 NIO 程序。
3. 开发工作量和难度都非常大：例如客户端面临断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常流的处理等等。
4. JDK NIO 的 Bug：例如臭名昭著的 Epoll Bug，它会导致 Selector 空轮询，最终导致 CPU 100%。直到 JDK 1.7 版本该问题仍旧存在，没有被根本解决。

tg:后面代码需要引入netty依赖

1. Netty 是由 JBOSS 提供的一个 Java 开源框架。Netty 提供异步的、基于事件驱动的网络应用程序框架，用以快速开发高性能、高可靠性的网络 IO 程序。
2. Netty 可以帮助你快速、简单的开发出一个网络应用，相当于简化和流程化了 NIO 的开发过程。
3. Netty 是目前最流行的 NIO 框架，Netty 在互联网领域、大数据分布式计算领域、游戏行业、通信行业等获得了广泛的应用，知名的 Elasticsearch 、Dubbo 框架内部都采用了 Netty。

线程模型基本介绍

1. 不同的线程模式，对程序的性能有很大影响，为了搞清Netty 线程模式我们来系统的讲解下 各个线程模式， 最后看看Netty 线程模型有什么优越性.
2. 目前存在的线程模型有：
   传统阻塞 I/O 服务模型
   Reactor 模式
3. 根据 Reactor 的数量和处理资源池线程的数量不同，有 3 种典型的实现
   • 单 Reactor 单线程；
   • 单 Reactor 多线程；
   • 主从 Reactor 多线程
4. Netty 线程模式(Netty 主要基于主从 Reactor 多线程模型做了一定的改进，其中主从Reactor 多线程模型有多个 Reactor)

传统阻塞 I/O 服务模型

Reactor 模式
Reactor 模式中 核心组成：

1. Reactor：Reactor 在一个单独的线程中运行，负责监听和分发事件，分发给适当的处理程序来对 IO 事件做出反应。 它就像公司的电话接线员，它接听来自客户的电话并将线路转移到适当的联系人；
2. Handlers：处理程序执行 I/O 事件要完成的实际事件，类似于客户想要与之交谈的公司中的实际官员。Reactor 通过调度适当的处理程序来响应 I/O 事件，处理程序执行非阻塞操作。

Reactor 模式分类：
根据 Reactor 的数量和处理资源池线程的数量不同，有 3 种典型的实现

1. 单 Reactor 单线程
2. 单 Reactor 多线程
3. 主从 Reactor 多线程

Netty快速入门实例-TCP服务

1. 实例要求：使用IDEA 创建Netty项目
2. Netty 服务器在 6668 端口监听，客户端能发送消息给服务器 “hello, 服务器~”
3. 服务器可以回复消息给客户端 “hello, 客户端~”

服务端

package Netty.simple.server;

import Netty.handler.NettyServerHandler;
import Netty.simple.handler.NettyServerHandlerSimple;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServerSimple {
    public static void main(String[] args) throws InterruptedException {
        // 创建eventloop boss worker
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        // 创建服务器启动对象 serverbootstrap
        ServerBootstrap bootstrap = new ServerBootstrap();
        try {
            bootstrap
                    .group(bossGroup, workerGroup)   // 设置两个线程组
                    .channel(NioServerSocketChannel.class)  // 用NioServerChannel 作为服务器通道实现
                    .option(ChannelOption.SO_BACKLOG, 124)  // 设置线程队列得到连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true)  // 设置保持活动连接状态
                    .childHandler(new ChannelInitializer() {  // 创建通道测试对象 匿名
                        @Override
                        // 给pipeline设置处理器
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new NettyServerHandlerSimple());
                        }
                    });
            System.out.println("服务端准备完成");
            // 绑定端口
            ChannelFuture cf = bootstrap.bind(7777).sync();
            // 对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端处理器

package Netty.simple.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;

public class NettyServerHandlerSimple extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("服务端读取线程:" + Thread.currentThread());
        System.out.println("ctx = " + ctx);
        System.out.println("channel 和 pipeline 关系");
        Channel channel = ctx.channel();
        ChannelPipeline pipeline = ctx.pipeline();
        // 对客户端消息处理
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发来消息:" + buf.toString(StandardCharsets.UTF_8));
        System.out.println("客户端地址:"+channel.localAddress());
    }
    // 读取数据完成处理

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 读取完成之后向服务端发送消息 编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,小可爱~~~", Charset.defaultCharset()));
    }
    // 异常处理

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

客户端

package Netty.simple.client;

import Netty.simple.handler.NettyClientHandlerSimple;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClientSimple {
    public static void main(String[] args) {
        // 创建worker
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 创建客户端启动对象
            Bootstrap bootstrap = new Bootstrap();
            // bootstrap 链式
            bootstrap
                    .group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyClientHandlerSimple());
                        }
                    });
            System.out.println("客户端准备完成");
            // 连接服务端 并获取future
            ChannelFuture cf = bootstrap.connect("localhost", 7777).sync();
            // 监听通道关闭
            cf.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

客户端处理器

package Netty.simple.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;

public class NettyClientHandlerSimple extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("ctx = " + ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("你好，我上线了", StandardCharsets.UTF_8));
    }
    // 接收服务端传来的消息
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务端回复你了>>" + buf.toString(Charset.defaultCharset()));
    }
    // 异常处理

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

异步模型

Future-Listener 机制

1. 当 Future 对象刚刚创建时，处于非完成状态，调用者可以通过返回的 ChannelFuture来获取操作执行的状态，注册监听函数来执行完成后的操作。
2. 常见有如下操作
   • 通过 isDone 方法来判断当前操作是否完成；
   • 通过 isSuccess 方法来判断已完成的当前操作是否成功；
   • 通过 getCause 方法来获取已完成的当前操作失败的原因；
   • 通过 isCancelled 方法来判断已完成的当前操作是否被取消；
   • 通过 addListener 方法来注册监听器，当操作已完成(isDone 方法返回完成)，将会通知指定的监听器；如果 Future 对象已完成，则通知指定的监听器。

任务队列中的 Task 有 3 种典型使用场景

1. 用户程序自定义的普通任务
2. 用户自定义定时任务
3. 非当前 Reactor 线程调用 Channel 的各种方法例如在推送系统的业务线程里面，根据用户的标识，找到对应的 Channel 引用，然后调用 Write 类方法向该用户推送消息，就会进入到这种场景。最终的 Write 会提交到任务队列中后被异步消费
   
   

快速入门实例-HTTP服务

1. 实例要求：使用IDEA 创建Netty项目
2. Netty 服务器在 6668 端口监听，浏览器发出请求"http://localhost:6668/ "
3. 服务器可以回复消息给客户端 "Hello! 我是服务器 5 ", 并对特定请求资源进行过滤. 4) 目的：Netty 可以做Http服务开发，并且理解Handler实例和客户端及其请求的关系.
   服务端

package Netty.simple.http;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class HttpServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new TestServerInitializer());
            System.out.println("服务器已准备就绪...");
            ChannelFuture cf = bootstrap.bind(7777).sync();
            cf.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

TestServerInitializer.java

package Netty.simple.http;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpContentDecoder;
import io.netty.handler.codec.http.HttpServerCodec;

public class TestServerInitializer extends ChannelInitializer {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        // 得到pipeline
        ChannelPipeline pipeline = socketChannel.pipeline();
        // netty自带http decode
        pipeline.addLast("MyHttpDecodeC", new HttpServerCodec());
        // 自定义handler
        pipeline.addLast("MyHttpServerHandler", new TestHttpServerHandler());
    }
}

TestHttpServerHandler.java

package Netty.simple.http;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;

import java.net.URI;
import java.nio.charset.Charset;

public class TestHttpServerHandler extends SimpleChannelInboundHandler {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject) throws Exception {
        if (httpObject instanceof HttpRequest) {
            System.out.println("pipeline hashcode " + channelHandlerContext.pipeline().hashCode() + "TestHttpServerHandler hashcode " + this.hashCode());
            System.out.println("httpObject 类型:" + httpObject.getClass());
            System.out.println("客户端地址:" + channelHandlerContext.channel().remoteAddress());
            HttpRequest request = (HttpRequest) httpObject;
            URI uri = new URI(request.uri());
            // 过滤图标资源
            if ("/favicon.ico".equals(uri.getPath())) {
                System.out.println("不是需要的资源");
                return;
            }
            ByteBuf buf = Unpooled.copiedBuffer("你好，小可爱~", Charset.defaultCharset());
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buf);
            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain;charset=utf-8");
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH, buf.readableBytes());
            channelHandlerContext.writeAndFlush(response);
        }
    }
}

Bootstrap、ServerBootstrap

1. Bootstrap 意思是引导，一个 Netty 应用通常由一个 Bootstrap 开始，主要作用是配置
   整个 Netty 程序，串联各个组件，Netty 中 Bootstrap 类是客户端程序的启动引导类，
   ServerBootstrap 是服务端启动引导类。
2. 常见的方法有
   • public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)，该方法用于服务器端，用来设置两个 EventLoop
   • public B group(EventLoopGroup group) ，该方法用于客户端，用来设置一个 EventLoop
   • public B channel(Class channelClass)，该方法用来设置一个服务器端的通道实现
   • public B option(ChannelOption option, T value)，用来给 ServerChannel 添加配置
   • public ServerBootstrap childOption(ChannelOption childOption, T value)，用来给接收到的
   通道添加配置
   • public ServerBootstrap childHandler(ChannelHandler childHandler)，该方法用来设置业务处理类
   （自定义的 handler）
   • public ChannelFuture bind(int inetPort) ，该方法用于服务器端，用来设置占用的端口号
   • public ChannelFuture connect(String inetHost, int inetPort) ，该方法用于客户端，用来连接服务器

Future、ChannelFuture

1. Netty 中所有的 IO 操作都是异步的，不能立刻得知消息是否被正确处理。但是可以过一会等它执行完成或者直接注册一个监听，具体的实现就是通过 Future 和ChannelFutures，他们可以注册一个监听，当操作执行成功或失败时监听会自动触发
   注册的监听事件
2. 常见的方法有
   • Channel channel()，返回当前正在进行 IO 操作的通道
   • ChannelFuture sync()，等待异步操作执行完毕

Channel

1. Netty 网络通信的组件，能够用于执行网络 I/O 操作
2. 通过Channel 可获得当前网络连接的通道的状态
3. 通过Channel 可获得 网络连接的配置参数 （例如接收缓冲区大小）
4. Channel 提供异步的网络 I/O 操作(如建立连接，读写，绑定端口)，异步调用意味着任何 I/O 调用都将立即返回，并且不保证在调用结束时所请求的 I/O 操作已完成
5. 调用立即返回一个 ChannelFuture 实例，通过注册监听器到 ChannelFuture 上，可以I/O 操作成功、失败或取消时回调通知调用方
6. 支持关联 I/O 操作与对应的处理程序
7. 不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应，常用的Channel 类型:
   • NioSocketChannel，异步的客户端 TCP Socket 连接。
   • NioServerSocketChannel，异步的服务器端 TCP Socket 连接。
   • NioDatagramChannel，异步的 UDP 连接。
   • NioSctpChannel，异步的客户端 Sctp 连接。
   • NioSctpServerChannel，异步的 Sctp 服务器端连接，这些通道涵盖了 UDP 和 TCP 网络 IO
   以及文件 IO。

Selector

1. Netty 基于 Selector 对象实现 I/O 多路复用，通过 Selector 一个线程可以监听多个连接的 Channel 事件。
2. 当向一个 Selector 中注册 Channel 后，Selector 内部的机制就可以自动不断地查询(Select) 这些注册的 Channel 是否有已就绪的 I/O 事件（例如可读，可写，网络连接完成等），这样程序就可以很简单地使用一个线程高效地管理多个 Channe。

ChannelHandler 及其实现类

链接: https://pan.baidu.com/s/1_17oheykJbdQlYMdsovrEA?pwd=hyrk 提取码: hyrk