前言：我们知道，Netty是基于NIO开发的一套框架，在学习Netty之前，我们先学习下Java NIO。
IO多路复用模型使用了Reactor设计模式，主要有三种实现：Reacotr单线程、Reactor多线程、Reactor主从模式。
在Reactor单线程模式中，所有客户端的请求处理都交给一个线程，串行化处理，效率较低。
在Reactor多线程模式中，acceptor线程负责接受客户端请求并将请求处理任务交给线程池，提升了请求处理速度。但是当client数量过多时，单线程就无法同时处理那么多的请求，造成瓶颈问题。
为了解决Reactor多线程请求转发瓶颈问题，Reactor主从模式将acceptor设计为线程池，用以处理客户端请求。
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); SetselectedKeys = selector.selectedKeys(); Iterator 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); } } }
服务端通讯序列图如下：
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); SetselectedKeys = selector.selectedKeys(); Iterator 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."); } }
客户端通讯序列图如下：
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数据读到一个内存块中，然后在这个块里读取，使用堆外内存可省去这一步骤）。
特性：
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);
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);
1. Netty准备知识：Java NIO
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有