网站推广的最终目的是什么,深圳前十大包装设计公司排名,wordpress文章添加忽略,安徽合肥做网站本次开始NIO网络编程#xff0c;之前已经说过BIO#xff0c;对于阻塞IO里面的问题一定有了清晰的认识#xff0c;在JDK1.4版本后#xff0c;提供了新的JAVA IO操作非阻塞API#xff0c;用意替换JAVA IO 和JAVA NetWorking相关的API。NIO其实有个名称叫new IO。(一)NIO① 介… 本次开始NIO网络编程之前已经说过BIO对于阻塞IO里面的问题一定有了清晰的认识在JDK1.4版本后提供了新的JAVA IO操作非阻塞API用意替换JAVA IO 和JAVA NetWorking相关的API。NIO其实有个名称叫new IO。(一)NIO① 介绍java.nio全称java non-blocking IO(实际上是 new io)是指JDK 1.4 及以上版本里提供的新api(New IO) 为所有的原始类型(boolean类型除外)提供缓存支持的数据容器使用它可以提供非阻塞式的高伸缩性网络。HTTP2.0使用了多路复用的技术做到同一个连接并发处理多个请求而且并发请求的数量比HTTP1.1大了好几个数量级。② 三大核心组件高性能网络编程的基础组件Buffer缓存区、Channel 通道、Selector 选择器。(二) Buffer缓存区① 介绍缓存区本质上是一个可以写入数据的内存块(类似数组)然后可以再次读取。此内存块包含在NIO Buffer 对象中该对象提供了一组方法可以更轻松地使用内存块。相比较直接对数组的操作。Buffer API 更加容易操作和管理。② 使用Buffer进行数据写入与读取需要进行如下四个步骤将数据写入缓冲区。调用buffer.flip()转换为读取模式。缓冲区读取数据。调用buffer.clear() 或 buffer.compact() 消除缓冲区③ Buffer工作原理BUffer三个重要属性通过完成了数组的封装。1.capacity 容量作为一个内存块Buffer具有一定的固定大小也称为【容量】。2.position 位置写入模式时代表写数据的位置。读取模式时代表读取数据的位置。3.limit 限制写入模式限制等于buffer的容量读取模式下limit等于写入的数据量。④ 源码import java.nio.ByteBuffer;import java.nio.IntBuffer;import java.nio.LongBuffer;public class BufferDemo {public static void main(String[] args) {// 构建一个byte字节缓冲区容量是4//堆内存 ByteBuffer byteBuffer ByteBuffer.allocate(4);//堆外内存// ByteBuffer byteBuffer ByteBuffer.allocateDirect(4);// 默认写入模式查看三个重要的指标 System.out.println(String.format(初始化capacity容量%s, position位置%s, limit限制%s, byteBuffer.capacity(), byteBuffer.position(), byteBuffer.limit()));// 写入2字节的数据 byteBuffer.put((byte) 1); byteBuffer.put((byte) 2); byteBuffer.put((byte) 3);// 再看数据 System.out.println(String.format(写入3字节后capacity容量%s, position位置%s, limit限制%s, byteBuffer.capacity(), byteBuffer.position(), byteBuffer.limit()));// 转换为读取模式(不调用flip方法也是可以读取数据的但是position记录读取的位置不对) System.out.println(#######开始读取); byteBuffer.flip();byte a byteBuffer.get(); System.out.println(a);byte b byteBuffer.get(); System.out.println(b); System.out.println(String.format(读取2字节数据后capacity容量%s, position位置%s, limit限制%s, byteBuffer.capacity(), byteBuffer.position(), byteBuffer.limit()));// 继续写入3字节此时读模式下limit3position2.继续写入只能覆盖写入一条数据// clear()方法清除整个缓冲区。compact()方法仅清除已阅读的数据。转为写入模式 byteBuffer.compact(); // buffer : 1 , 3 byteBuffer.put((byte) 3); byteBuffer.put((byte) 4); byteBuffer.put((byte) 5); System.out.println(String.format(最终的情况capacity容量%s, position位置%s, limit限制%s, byteBuffer.capacity(), byteBuffer.position(), byteBuffer.limit()));// rewind() 重置position为0// mark() 标记position的位置// reset() 重置position为上次mark()标记的位置 }}⑤ ByteBuffer 内存类型ByteBuffer 为性能关键型代码提供了直接内存(direct堆外)和非直接内存(heap堆)两种实现堆外内存获取的方式ByteBuffer directBytebuffer ByteBuffer.allocateDirect(noBytes);好处进行网络IO 或者 文件IO时比heapBuffer 少一次拷贝(file/socket —— OS memory —— jvm heap )GC会移动对象内存在写file 或 socket的过程中JVM的实现中会先把数据复制到堆外在进行写入。GC范围之外降低GC压力但实现了自动管理。DirectByteBuffer 中 有一个Cleaner 对象(PhantomReference) Cleaner被GC前会执行clean 方法触发DirectByteBuffer 中定义Deallocator建议性能确实可观的时候才去使用分配给大型长寿命(网络传输文件读写场景)通过虚拟机参数MaxDirectMemorySize限制大小防止耗尽整个机器的内存在JVM之外的内存无法监控。(三)Channel 通道① 介绍Channel的API 涵盖了UDP、TCP网络和文件IOFileChannelDatagramChannelSocketChannelServerSocketChannel。② 和标准IO Stream操作的区别在一个通道内进行读取和写入stream通常是单向的(input 或 output)可以非堵塞读取和写入通道通道中读取或写入缓冲区。③ SocketChannelSocketChannel用于建立TCP网络连接类似java.net.Socket。有两种创建socketChannel形式1.客户端主动发起和服务器的连接2.服务器获取的新连接write写在尚未写入任何内容时可能就返回了。需要在循环中调用write()read读read() 方法可能直接返回而根本不读取任何数据根据返回的int值判断读取了多少字节。④ ServerSocketChannelServerSocketChannel 可能监听新建立的TCP连接通道类似ServerSocket。ServerSocketChannel.accepta()如果该通道处于飞度赛模式那么如何没有挂起的连接该方法将立即返回null。必须检查返回的SocketChannel是否为null。⑤ 源码import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.SocketChannel;import java.util.Scanner;public class NIOClient {public static void main(String[] args) throws Exception { SocketChannel socketChannel SocketChannel.open(); socketChannel.configureBlocking(false); socketChannel.connect(new InetSocketAddress(127.0.0.1, 8080));while (!socketChannel.finishConnect()) {// 没连接上,则一直等待 Thread.yield(); } Scanner scanner new Scanner(System.in); System.out.println(请输入);// 发送内容 String msg scanner.nextLine(); ByteBuffer buffer ByteBuffer.wrap(msg.getBytes());while (buffer.hasRemaining()) { socketChannel.write(buffer); }// 读取响应 System.out.println(收到服务端响应:); ByteBuffer requestBuffer ByteBuffer.allocate(1024);while (socketChannel.isOpen() socketChannel.read(requestBuffer) ! -1) {// 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)if (requestBuffer.position() 0) break; } requestBuffer.flip();byte[] content new byte[requestBuffer.limit()]; requestBuffer.get(content); System.out.println(new String(content)); scanner.close(); socketChannel.close(); }}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.ArrayList;import java.util.Iterator;/** * 直接基于非阻塞的写法,一个线程处理轮询所有请求 */public class NIOServer1 {/** * 已经建立连接的集合 */private static ArrayList channels new ArrayList();public static void main(String[] args) throws Exception {// 创建网络服务端 ServerSocketChannel serverSocketChannel ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false); // 设置为非阻塞模式 serverSocketChannel.socket().bind(new InetSocketAddress(8080)); // 绑定端口 System.out.println(启动成功);while (true) { SocketChannel socketChannel serverSocketChannel.accept(); // 获取新tcp连接通道// tcp请求 读取/响应if (socketChannel ! null) { System.out.println(收到新连接 : socketChannel.getRemoteAddress()); socketChannel.configureBlocking(false); // 默认是阻塞的,一定要设置为非阻塞 channels.add(socketChannel); } else {// 没有新连接的情况下,就去处理现有连接的数据,处理完的就删除掉 Iterator iterator channels.iterator();while (iterator.hasNext()) { SocketChannel ch iterator.next();try { ByteBuffer requestBuffer ByteBuffer.allocate(1024);if (ch.read(requestBuffer) 0) {// 等于0,代表这个通道没有数据需要处理,那就待会再处理continue; }while (ch.isOpen() ch.read(requestBuffer) ! -1) {// 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)if (requestBuffer.position() 0) break; }if(requestBuffer.position() 0) continue; // 如果没数据了, 则不继续后面的处理 requestBuffer.flip();byte[] content new byte[requestBuffer.limit()]; requestBuffer.get(content); System.out.println(new String(content)); System.out.println(收到数据,来自 ch.getRemoteAddress());// 响应结果 200 String response HTTP/1.1 200 OK\r\n Content-Length: 11\r\n\r\n Hello World; ByteBuffer buffer ByteBuffer.wrap(response.getBytes());while (buffer.hasRemaining()) { ch.write(buffer); } iterator.remove(); } catch (IOException e) { e.printStackTrace(); iterator.remove(); } } } }// 用到了非阻塞的API, 再设计上,和BIO可以有很大的不同// 问题: 轮询通道的方式,低效,浪费CPU }}(四)Select选择器① 介绍Selector 是一个Java NIO 组件可以检查一个或多个NIO通道并确定哪些通道已准备好进行读取或写入实现单个线程可以管理多个通道从而管理或多个网络连接。② selector 监听多个 channel的不同事件Connect 连接(SelectionKey.OP_CONNECT)Accept 准备就绪(OP_ACCEPT)Read 读取(OP_READ)Write 写入(OP_WRITE)③ selector 选择器一个线程处理多个通道的核心概念理解事件驱动机制。非堵塞的网络通道下开发者通过Selector注册对于通道感兴趣的事件类型线程通过监听事件来触发响应的代码执行(最底层hi操作系统的多路复用机制)④ 源码import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.*;import java.util.Iterator;import java.util.Set;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/** * 结合Selector实现的非阻塞服务端(放弃对channel的轮询,借助消息通知机制) */public class NIOServerV2 {public static void main(String[] args) throws Exception {// 1. 创建网络服务端ServerSocketChannel ServerSocketChannel serverSocketChannel ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false); // 设置为非阻塞模式// 2. 构建一个Selector选择器,并且将channel注册上去 Selector selector Selector.open(); SelectionKey selectionKey serverSocketChannel.register(selector, 0, serverSocketChannel);// 将serverSocketChannel注册到selector selectionKey.interestOps(SelectionKey.OP_ACCEPT); // 对serverSocketChannel上面的accept事件感兴趣(serverSocketChannel只能支持accept操作)// 3. 绑定端口 serverSocketChannel.socket().bind(new InetSocketAddress(8080)); System.out.println(启动成功);while (true) {// 不再轮询通道,改用下面轮询事件的方式.select方法有阻塞效果,直到有事件通知才会有返回 selector.select();// 获取事件 Set selectionKeys selector.selectedKeys();// 遍历查询结果e Iterator iter selectionKeys.iterator();while (iter.hasNext()) {// 被封装的查询结果 SelectionKey key iter.next(); iter.remove();// 关注 Read 和 Accept两个事件if (key.isAcceptable()) { ServerSocketChannel server (ServerSocketChannel) key.attachment();// 将拿到的客户端连接通道,注册到selector上面 SocketChannel clientSocketChannel server.accept(); // mainReactor 轮询accept clientSocketChannel.configureBlocking(false); clientSocketChannel.register(selector, SelectionKey.OP_READ, clientSocketChannel); System.out.println(收到新连接 : clientSocketChannel.getRemoteAddress()); }if (key.isReadable()) { SocketChannel socketChannel (SocketChannel) key.attachment();try { ByteBuffer requestBuffer ByteBuffer.allocate(1024);while (socketChannel.isOpen() socketChannel.read(requestBuffer) ! -1) {// 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)if (requestBuffer.position() 0) break; }if(requestBuffer.position() 0) continue; // 如果没数据了, 则不继续后面的处理 requestBuffer.flip();byte[] content new byte[requestBuffer.limit()]; requestBuffer.get(content); System.out.println(new String(content)); System.out.println(收到数据,来自 socketChannel.getRemoteAddress());// TODO 业务操作 数据库 接口调用等等// 响应结果 200 String response HTTP/1.1 200 OK\r\n Content-Length: 11\r\n\r\n Hello World; ByteBuffer buffer ByteBuffer.wrap(response.getBytes());while (buffer.hasRemaining()) { socketChannel.write(buffer); } } catch (IOException e) {// e.printStackTrace(); key.cancel(); // 取消事件订阅 } } } selector.selectNow(); }// 问题: 此处一个selector监听所有事件,一个线程处理所有请求事件. 会成为瓶颈! 要有多线程的运用 }}⑤ NIO 和 BIO 的区别⑥ NIO Reactor的方式import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.*;import java.util.Iterator;import java.util.Random;import java.util.Set;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.FutureTask;import java.util.concurrent.LinkedBlockingQueue;import java.util.concurrent.atomic.AtomicInteger;/** * NIO selector 多路复用reactor线程模型 */public class NIOServerV3 {/** 处理业务操作的线程 */private static ExecutorService workPool Executors.newCachedThreadPool();/** * 封装了selector.select()等事件轮询的代码 */abstract class ReactorThread extends Thread { Selector selector; LinkedBlockingQueue taskQueue new LinkedBlockingQueue();/** * Selector监听到有事件后,调用这个方法 */public abstract void handler(SelectableChannel channel) throws Exception;private ReactorThread() throws IOException { selector Selector.open(); }volatile boolean running false;Overridepublic void run() {// 轮询Selector事件while (running) {try {// 执行队列中的任务 Runnable task;while ((task taskQueue.poll()) ! null) { task.run(); } selector.select(1000);// 获取查询结果 Set selected selector.selectedKeys();// 遍历查询结果 Iterator iter selected.iterator();while (iter.hasNext()) {// 被封装的查询结果 SelectionKey key iter.next(); iter.remove();int readyOps key.readyOps();// 关注 Read 和 Accept两个事件if ((readyOps (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) ! 0 || readyOps 0) {try { SelectableChannel channel (SelectableChannel) key.attachment(); channel.configureBlocking(false); handler(channel);if (!channel.isOpen()) { key.cancel(); // 如果关闭了,就取消这个KEY的订阅 } } catch (Exception ex) { key.cancel(); // 如果有异常,就取消这个KEY的订阅 } } } selector.selectNow(); } catch (IOException e) { e.printStackTrace(); } } }private SelectionKey register(SelectableChannel channel) throws Exception {// 为什么register要以任务提交的形式让reactor线程去处理// 因为线程在执行channel注册到selector的过程中会和调用selector.select()方法的线程争用同一把锁// 而select()方法实在eventLoop中通过while循环调用的争抢的可能性很高为了让register能更快的执行就放到同一个线程来处理 FutureTask futureTask new FutureTask(() - channel.register(selector, 0, channel)); taskQueue.add(futureTask);return futureTask.get(); }private void doStart() {if (!running) { running true; start(); } } }private ServerSocketChannel serverSocketChannel;// 1、创建多个线程 - accept处理reactor线程 (accept线程)private ReactorThread[] mainReactorThreads new ReactorThread[1];// 2、创建多个线程 - io处理reactor线程 (I/O线程)private ReactorThread[] subReactorThreads new ReactorThread[8];/** * 初始化线程组 */private void newGroup() throws IOException {// 创建IO线程,负责处理客户端连接以后socketChannel的IO读写for (int i 0; i subReactorThreads.length; i) { subReactorThreads[i] new ReactorThread() {Overridepublic void handler(SelectableChannel channel) throws IOException {// work线程只负责处理IO处理不处理accept事件 SocketChannel ch (SocketChannel) channel; ByteBuffer requestBuffer ByteBuffer.allocate(1024);while (ch.isOpen() ch.read(requestBuffer) ! -1) {// 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)if (requestBuffer.position() 0) break; }if (requestBuffer.position() 0) return; // 如果没数据了, 则不继续后面的处理 requestBuffer.flip();byte[] content new byte[requestBuffer.limit()]; requestBuffer.get(content); System.out.println(new String(content)); System.out.println(Thread.currentThread().getName() 收到数据,来自 ch.getRemoteAddress());// TODO 业务操作 数据库、接口... workPool.submit(() - { });// 响应结果 200 String response HTTP/1.1 200 OK\r\n Content-Length: 11\r\n\r\n Hello World; ByteBuffer buffer ByteBuffer.wrap(response.getBytes());while (buffer.hasRemaining()) { ch.write(buffer); } } }; }// 创建mainReactor线程, 只负责处理serverSocketChannelfor (int i 0; i mainReactorThreads.length; i) { mainReactorThreads[i] new ReactorThread() { AtomicInteger incr new AtomicInteger(0);Overridepublic void handler(SelectableChannel channel) throws Exception {// 只做请求分发不做具体的数据读取 ServerSocketChannel ch (ServerSocketChannel) channel; SocketChannel socketChannel ch.accept(); socketChannel.configureBlocking(false);// 收到连接建立的通知之后分发给I/O线程继续去读取数据int index incr.getAndIncrement() % subReactorThreads.length; ReactorThread workEventLoop subReactorThreads[index]; workEventLoop.doStart(); SelectionKey selectionKey workEventLoop.register(socketChannel); selectionKey.interestOps(SelectionKey.OP_READ); System.out.println(Thread.currentThread().getName() 收到新连接 : socketChannel.getRemoteAddress()); } }; } }/** * 初始化channel,并且绑定一个eventLoop线程 * * throws IOException IO异常 */private void initAndRegister() throws Exception {// 1、 创建ServerSocketChannel serverSocketChannel ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false);// 2、 将serverSocketChannel注册到selectorint index new Random().nextInt(mainReactorThreads.length); mainReactorThreads[index].doStart(); SelectionKey selectionKey mainReactorThreads[index].register(serverSocketChannel); selectionKey.interestOps(SelectionKey.OP_ACCEPT); }/** * 绑定端口 * * throws IOException IO异常 */private void bind() throws IOException {// 1、 正式绑定端口对外服务 serverSocketChannel.bind(new InetSocketAddress(8080)); System.out.println(启动完成端口8080); }public static void main(String[] args) throws Exception { NIOServerV3 nioServerV3 new NIOServerV3(); nioServerV3.newGroup(); // 1、 创建main和sub两组线程 nioServerV3.initAndRegister(); // 2、 创建serverSocketChannel注册到mainReactor线程上的selector上 nioServerV3.bind(); // 3、 为serverSocketChannel绑定端口 }}PSNIO为开发者提供了功能丰富及强大的IO处理API但是在应用开发的过程中直接使用JDK提供的API比较繁琐而且要想将性能进行提升光有NIO还是不够的还需要将多线程技术与之结合起来。因为网络编程本身的复杂性以及JDK API开发的使用难度较高所以开源社区中涌出来很多的JDK NIO进行封装了增强后的网络编程框架例如Netty、Mina等。
