1. Netty准备知识:Java NIO

前言:我们知道,Netty是基于NIO开发的一套框架,在学习Netty之前,我们先学习下Java NIO。

一、IO多路复用模型

  IO多路复用模型使用了Reactor设计模式,主要有三种实现:Reacotr单线程、Reactor多线程、Reactor主从模式。

1. Reactor单线程

  在Reactor单线程模式中,所有客户端的请求处理都交给一个线程,串行化处理,效率较低。

2. Reactor多线程

  在Reactor多线程模式中,acceptor线程负责接受客户端请求并将请求处理任务交给线程池,提升了请求处理速度。但是当client数量过多时,单线程就无法同时处理那么多的请求,造成瓶颈问题。

3. Reactor主从模式

  为了解决Reactor多线程请求转发瓶颈问题,Reactor主从模式将acceptor设计为线程池,用以处理客户端请求。

二、NIO使用示例

1. NIO服务端通讯示例

public class TimeServer {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用默认值
            }
        }
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
        new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
    }
}

public class MultiplexerTimeServer implements Runnable {
    private Selector selector;
    private ServerSocketChannel servChannel;
    private volatile boolean stop;//保证线程可见(volitile关键字)

    /**
     * 初始化多路复用器、绑定监听端口
     */
    public MultiplexerTimeServer(int port) {
        try {
            selector = Selector.open();//创建多路复用器
            servChannel = ServerSocketChannel.open();//打开ServerSocketChannel,用于监听客户端链接
            servChannel.configureBlocking(false);//设置非阻塞
            servChannel.socket().bind(new InetSocketAddress(port), 1024);//绑定端口
            servChannel.register(selector, SelectionKey.OP_ACCEPT);//注册监听(监听ACCEPT事件)
            System.out.println("The time server is start in port : " + port);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    public void stop() {
        this.stop = true;
    }

    /**
     * 无限轮询准备就绪的key,并对其进行处理
     */
    @Override
    public void run() {
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null)
                                key.channel().close();
                        }
                    }
                }
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }

        // 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
        if (selector != null)
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
    }

    private void handleInput(SelectionKey key) throws IOException {
        if (key.isValid()) {
            // 处理新接入的请求消息
            if (key.isAcceptable()) {
                // Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                // Add the new connection to the selector
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                // Read the data
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("The time server receive order : " + body);
                    String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)
                            ? new java.util.Date(System.currentTimeMillis()).toString()
                            : "BAD ORDER";
                    doWrite(sc, currentTime);
                } else if (readBytes < 0) {
                    // 对端链路关闭
                    key.cancel();
                    sc.close();
                } else {
                    // 读到0字节,忽略
                }
            }
        }
    }

    private void doWrite(SocketChannel channel, String response) throws IOException {
        if (response != null && response.trim().length() > 0) {
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }
}

服务端通讯序列图如下:

2. 客户端通讯示例

public class TimeClient {
    public static void main(String[] args) {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用默认值
            }
        }
        new Thread(new TimeClientHandle("127.0.0.1", port), "TimeClient-001").start();
    }
}

public class TimeClientHandle implements Runnable {
    private String host;
    private int port;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;

    public TimeClientHandle(String host, int port) {
        this.host = host == null ? "127.0.0.1" : host;
        this.port = port;
        try {
            selector = Selector.open();// 创建多路复用器
            socketChannel = SocketChannel.open();// 打开SocketChannel
            socketChannel.configureBlocking(false);// 设置非阻塞模式
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    /**
     * 创建连接,无线循环准备好的key并对其进行处理
     */
    @Override
    public void run() {
        try {
            // 创建连接
            doConnect();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null)
                                key.channel().close();
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                System.exit(1);
            }
        }

        // 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }

    private void handleInput(SelectionKey key) throws IOException {
        if (key.isValid()) {
            // 判断是否连接成功
            SocketChannel sc = (SocketChannel) key.channel();
            if (key.isConnectable()) {
                if (sc.finishConnect()) {
                    sc.register(selector, SelectionKey.OP_READ);//注册READ事件
                    doWrite(sc);
                } else
                    System.exit(1);// 连接失败,进程退出
            }
            if (key.isReadable()) {
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("Now is : " + body);
                    this.stop = true;
                } else if (readBytes < 0) {
                    // 对端链路关闭
                    key.cancel();
                    sc.close();
                } else {
                    // 读到0字节,忽略
                }
            }
        }
    }

    private void doConnect() throws IOException {
        // 如果直接连接成功,则注册到多路复用器上,发送请求消息,读应答
        if (socketChannel.connect(new InetSocketAddress(host, port))) {
            socketChannel.register(selector, SelectionKey.OP_READ);
            doWrite(socketChannel);
        } else
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
    }

    private void doWrite(SocketChannel sc) throws IOException {
        byte[] req = "QUERY TIME ORDER".getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
        writeBuffer.put(req);
        writeBuffer.flip();
        sc.write(writeBuffer);
        if (!writeBuffer.hasRemaining())
            System.out.println("Send order 2 server succeed.");
    }
}

客户端通讯序列图如下:

三、NIO类库介绍

1. 缓冲区Buffer

  Buffer是一个对象,它包含一些要写入或读出的数据。在NIO中,所有的数据都是在缓冲区处理的,读数据时,直接读进缓冲区,写数据时,直接写在缓冲区。

  Buffer类库继承关系:

我们最常用的就是ByteBuffer,这里介绍下ByteBuffer的几个参数及使用方法:

capacity 数组容量,创建后不可变
limit 当写数据到buffer中时,limit一般和capacity相等,当读数据时,limit代表buffer中有效数据的长度
position 位置,下一次要被读或被写的位置
mark 标记,调用mark()来设置mark=position,调用reset()可以让position恢复到标记位置
clear() 令position=0;limit=capacity;mark=-1;  但是不清除byte数组内容
reset() 把position设置成mark的值,相当于之前做过一个标记,现在要退回到之前标记的地方
flip() 令limit=position;position=0
allocate(int capavity) 从堆空间中分配一个容量大小为capacity的byte数组作为缓冲区的byte数据存储器
allocateDirect(int capacity) 在堆外内存中分配一个容量大小为capacity的byte数组作为缓冲区的byte数据存储器
wrap(byte[] array) 将byte数组包装成ByteBuffer

tips:ByteBuffer有两个比较重要的实现,HeapByteBuffer 和 DirectByteBuffer。

  HeapByteBuffer:在堆内申请的内存,利于维护

  DirectByteBuffer:在堆外申请的内存,实现零拷贝,提升数据操作速度(外部读取JVM堆中数据是先把JVM数据读到一个内存块中,然后在这个块里读取,使用堆外内存可省去这一步骤)。

2. 通道Channel

特性:

1. 既可以从通道中读取数据,又可以写数据到通道。但流的读写通常是单向的。

2. 通道可以异步地读写。

3. 通道中的数据总是要先读到一个Buffer,或者总是要从一个Buffer中写入。

重要实现:

FileChannel:从文件中读写数据

DatagramChannel:通过UDP读写网络中的数据,因为UDP是无连接的网络协议,所以不能像其它通道那样读取和写入,它发送和接收的是数据包

SocketChannel:通过TCP读写网络中的数据

ServerSocketChannel:监听新进来的TCP连接,对每一个连接创建一个SocketChannel

  示例(这里只举例了FileChannel 和 DatagramChannel,其他两种看上面第二部分):

//FileChannel示例:

RandomAccessFile aFile = new RandomAccessFile("data/nio-data.txt", "rw");

FileChannel inChannel = aFile.getChannel();
// 设置1MB的缓冲区

ByteBuffer buf = ByteBuffer.allocate(1024);
// 读取数据到buf中,并返回字节数

int bytesRead = inChannel.read(buf);

while (bytesRead != -1) {

    System.out.println("Read " + bytesRead);

    buf.flip(); // 重设缓冲区 postion = 0 ,limit = 原本position

    while(buf.hasRemaining()){ //缓冲区中是否还有内容
        // 或者使用buf.get(bytes),将数据读进字节数组中

        System.out.print((char) buf.get());

    }

    buf.clear();//清空缓存区
    // 缓冲区只有1MB大小,需要循环读取

    bytesRead = inChannel.read(buf);

}
aFile.close();

//DataGramChannel:

// 打开连接
DatagramChannel channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(port));

// buffer接收channel的数据
channel.receive(buf);//如果buffer容不下收到的数据,多出的数据将被抛弃

// 发送数据
channel.send(buf, new InetSocketAddress("baidu.com", 80));

// 连接到特定地址:由于UDP是无连接的,连接到特定的地址并不会像TCP通道那样创建一个真正的连接,而是锁住DatagramChannel,让其只能从特定地址收发数据
channel.connect(new InetSocketAddress("baidu.com", 80));
连接后,可以使用read()和write()方法,但数据传送无保证
channel.read(buf);
channel.write(buf);

Channel示例

3. 多路复用器Selector

  Selector不断轮询注册在其上的Channel,如果某个Channel上有新的TCP连接、读、写操作,这个Channel就处于就绪状态,会被Selector轮询出来,然后通过SelectionKey就可以获取到就绪Channel集合,进行后续的IO操作。

  一些方法:

// 1. 创建Selector
Selector selector = Selector.open();

// 2. 注册通道
channel.configureBlocking(false);//channel必须处于非阻塞状态
SelectionKey key = channel.register(selector, Selector.OP_READ);

int select():阻塞到至少有一个通道在你注册的事件上就绪了,返回值为多少通道已就绪
int select(long timeout):同上,最长阻塞timeout毫秒
int selectNow():不阻塞,不管什么通道都立即返回
Set selectedKeys = selector.selectedKeys():返回已就绪的通道的SelectedKey

我们获取到的是SelectionKey,该对象中包含了一些有价值的属性:

insterest集合:OP_CONNECT、OP_ACCEPT、OP_READ、OP_WRITE
    int interestSet = selectionKey.interestOps();
    boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT;
    boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
    boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
    boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;

ready集合:已准备就绪的操作的集合(你注册了监听connect,那么除了connect其它都是false),在一次选择(Selection)之后,会首先访问这个ready set
     int readySet = selectionKey.readyOps();
     boolean isAccept = selectionKey.isAcceptable();
     boolean isConnect = selectionKey.isConnectable();
     boolean isReadable = selectionKey.isReadable();
     boolean isWritable = selectionKey.isWritable();

channel:Channel  channel  = selectionKey.channel();
selector:Selector selector = selectionKey.selector();
附加对象(可选):SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

原文地址:https://www.cnblogs.com/lovezmc/p/11547841.html

时间: 2024-11-07 06:15:31

1. Netty准备知识:Java NIO的相关文章

5. 彤哥说netty系列之Java NIO核心组件之Channel

你好,我是彤哥,本篇是netty系列的第五篇. 简介 上一章我们一起学习了如何使用Java原生NIO实现群聊系统,这章我们一起来看看Java NIO的核心组件之一--Channel. 思维转变 首先,我想说的最重要的一个点是,学习NIO思维一定要从BIO那种一个连接一个线程的模式转变成多个连接(Channel)共用一个线程来处理的这种思维. 1个Connection = 1个Socket = 1个Channel,这几个概念可以看作是等价的,都表示一个连接,只不过是用在不同的场景中. 如果单从阻塞

6. 彤哥说netty系列之Java NIO核心组件之Buffer

--日拱一卒,不期而至! 你好,我是彤哥,本篇是netty系列的第六篇. 简介 上一章我们一起学习了Java NIO的核心组件Channel,它可以看作是实体与实体之间的连接,而且需要与Buffer交互,这一章我们就来学习一下Buffer的特性. 概念 Buffer用于与Channel交互时使用,通过上一章的学习我们知道,数据从Channel读取到Buffer,或者从Buffer写入Channel. Buffer本质上是一个内存块,可以向里面写入数据,或者从里面读取数据,在Java中它被包装成了

7. 彤哥说netty系列之Java NIO核心组件之Selector

<p align="right">--日拱一卒,不期而至!</p> 你好,我是彤哥,本篇是netty系列的第七篇. 简介 上一章我们一起学习了Java NIO的核心组件Buffer,它通常跟Channel一起使用,但是它们在网络IO中又该如何使用呢,今天我们将一起学习另一个NIO核心组件--Selector,没有它可以说就干不起来网络IO. 概念 我们先来看两段Selector的注释,见类java.nio.channels.Selector. 注释I A mul

JAVA NIO 类库的异步通信框架netty和mina

Netty 和 Mina 我究竟该选择哪个? 根据我的经验,无论选择哪个,都是个正确的选择.两者各有千秋,Netty 在内存管理方面更胜一筹,综合性能也更优.但是,API 变更的管理和兼容性做的不是太好.相比于 Netty,Mina 的前向兼容性.内聚的可维护性功能更多,例如 JMX 的集成.性能统计.状态机等. Netty 是业界最流行的 NIO 框架之一,它的健壮性.功能.性能.可定制性和可扩展性在同类框架中都是首屈一指的,它已经得到成百上千的商用项目验证,例如 Hadoop 的 RPC 框

Java NIO 基础知识

前言 前言部分是科普,读者可自行选择是否阅读这部分内容. 为什么我们需要关心 NIO?我想很多业务猿都会有这个疑问. 我在工作的前两年对这个问题也很不解,因为那个时候我认为自己已经非常熟悉 IO 操作了,读写文件什么的都非常溜了,IO 包无非就是 File.RandomAccessFile.字节流.字符流这些,感觉没什么好纠结的.最混乱的当属 InputStream/OutputStream 一大堆的类不知道谁是谁,不过了解了装饰者模式以后,也都轻松破解了. 在 Java 领域,一般性的文件操作

3. 彤哥说netty系列之Java BIO NIO AIO进化史.md

你好,我是彤哥,本篇是netty系列的第三篇. 欢迎来我的公从号彤哥读源码系统地学习源码&架构的知识. 先说两个事 (1)上周五的那篇文章发重复了,是定时任务设置错误导致,给大家带来干扰,这里说声抱歉. (2)之前的问卷调查结果出来了,认为先讲案例的票数较多,所以后面的文章都是先讲案例,再以案例展开讲解组件. 简介 上一章我们介绍了IO的五种模型,实际上Java只支持其中的三种,即BIO/NIO/AIO. 本文将介绍Java中这三种IO的进化史,并从使用的角度剖析它们背后的故事. Java BI

学习 java netty (一) -- java nio

前言:最近在研究java netty这个网络框架,第一篇先介绍java的nio. java nio在jdk1.4引入,其实也算比较早的了,主要引入非阻塞io和io多路复用.内部基于reactor模式. nio核心: - buffer - channel - selector buffer: 类似网络编程中的缓冲区,有 ByteBuffer 字节 CharBuffer 字符 IntBuffer DoubleBuffer- 常用的有ByteBuffer和CharBuffer java nio buf

Netty 中 IOException: Connection reset by peer 与 java.nio.channels.ClosedChannelException: null

最近发现系统中出现了很多 IOException: Connection reset by peer 与 ClosedChannelException: null 深入看了看代码, 做了些测试, 发现 Connection reset 会在客户端不知道 channel 被关闭的情况下, 触发了 eventloop 的 unsafe.read() 操作抛出 而 ClosedChannelException 一般是由 Netty 主动抛出的, 在 AbstractChannel 以及 SSLHand

JAVA NIO buffer (知识三)

java nio 里的buffer是缓存数据,通常缓冲区是一个数组,字节数组,也可以是别的类型.最常用的就是bytebuffer, 还有一些其它的类型: charbuffer, shortbuffer, intbuffer, longbuffer, floatbuffer, doublebufer. 一开始在知识(一)里写到,想要用nio读取数据,都是从channel读取到buffer.然后应用从buffer读取数据,同样写数据也是,先把数据写到buffer中,然后读道channel中. 基本上