1. Netty准备知识:Java NIO

Posted lovezmc

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了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_ACCEPTboolean 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);

 

以上是关于1. Netty准备知识:Java NIO的主要内容,如果未能解决你的问题,请参考以下文章

「Java面试必看」Netty面试必问的知识点整理

Netty知识点总结

Java中的NIO基础知识

Java面试基础知识,java语言编程入门

杂谈 : 聊一聊NIO

Netty 核心知识点