NIO之ByteBuffer與Channel

傳統IO:byte[] <= inputStream <= 文件 => outputStream => byte[] NIO:文件 => inputChannel <=> buffer <=> outputChannel => 文件文件 <= inputChannel <=> outputChannel => 文件

• 文件復制, 并測試ByteBuffer常用API
  position: 當前指針位置; limit: 當前內容的最大位置(例如: buffer內容為"hello", 容量為20, limit就是5); capacity: 最大容量

/*** 測試Buffer的position, limit, capacity, clear, flip* @author regotto*/ public class NioTest1 {/*** 下面的代碼中, clear與flip效果一樣.* 沒有clear,flip,position的位置將會一直等于limit, 根據各屬性之間的大小關系, position一定不會大于limit, * 所以在下面的read中, 讀取到的值將一直都是0(代表當前讀取到的位置), read一直不會等于-1, 代碼出現死循環* @param args* @throws Exception*/public static void main(String[] args) throws Exception {FileChannel inputChannel = new FileInputStream("input.txt").getChannel();FileChannel outputChannel = new FileOutputStream("output.txt").getChannel();ByteBuffer buffer = ByteBuffer.allocate(4);while(true){//此處不進行clear, 此時position還是處于limit的位置, read將一直保持當前位置出現//死循環buffer.clear();int read = inputChannel.read(buffer);System.out.println("read: " + read);if (-1 == read){break;}//重置bufferbuffer.flip();outputChannel.write(buffer);}inputChannel.close();outputChannel.close();} }

clear, flip源碼如下:

public final Buffer clear() {position = 0;limit = capacity;mark = -1;return this;}public final Buffer flip() {limit = position;position = 0;mark = -1;return this;}

• DirectByteBuffer
  ByteBuffer.allocate(1024) => HeapByteBuffer, 內部使用的就是byte[], 底層源碼如下

public static ByteBuffer allocate(int capacity) {if (capacity < 0)throw new IllegalArgumentException();return new HeapByteBuffer(capacity, capacity);}//HeapByteBuffer extends ByteBuffer//ByteBuffer構造函數如下:ByteBuffer(int mark, int pos, int lim, int cap, // package-privatebyte[] hb, int offset){super(mark, pos, lim, cap);this.hb = hb;this.offset = offset;}

HeapByteBuffer位于JVM堆空間, 當使用HeapByteBuffer進行內容復制時, 存在2個復制過程: 應用程序 => 應用程序緩沖區 => 內核緩沖區 => 文件; 這種情況下, 2個復制過程存在一定的性能問題;

ByteBuffer.allocateDirect(1024) => DirectByteBuffer, DirectByteBuffer使用native方法創建數組, 數組不再位于JVM的Heap中, 而是位于內核內存中, 這樣就避免了一次數據拷貝(拷貝的原因在于JVM中數據的地址會改變, 在GC下): 應用程序緩沖區 -> 內核緩沖區; 所謂的零拷貝, 加快速度, C/C++開辟的數組空間都是位于內核緩沖區
DirectByteBuffer底層源碼:

public static ByteBuffer allocateDirect(int capacity) {return new DirectByteBuffer(capacity);}//unsafe中底層內存分配base = unsafe.allocateMemory(size);public native long allocateMemory(long var1);

使用DirectByteBuffer進行文件復制:

/*** 測試DirectByteBuffer* @author regotto*/ public class NioTest2 {public static void main(String[] args) throws Exception {FileChannel inputChannel = new FileInputStream("input.txt").getChannel();FileChannel outputChannel = new FileOutputStream("output.txt").getChannel();ByteBuffer buffer = ByteBuffer.allocateDirect(4);while(true){buffer.clear();int read = inputChannel.read(buffer);System.out.println("read: " + read);if (-1 == read){break;}buffer.flip();outputChannel.write(buffer);// buffer.flip();}inputChannel.close();outputChannel.close();}/*** 進行文件復制*/public void test() throws Exception {FileChannel fisChannel = new FileInputStream("text1.txt").getChannel();FileChannel fosChannel = new FileOutputStream("text2.txt").getChannel();//transferTo與transferFrom效果一樣fisChannel.transferTo(0, fisChannel.size(), fosChannel);fisChannel.close();fosChannel.close();} }

• 使用堆外內存進行文件內容復制(使用塊內存提高性能)

/*** 測試MappedByteBuffer* 使用堆外內存對文件內容進行修改* @author regotto*/ public class NioTest3 {public static void main(String[] args) throws Exception{//下面的0, 4代表從0號位開始,將4個大小的空間映射到堆外內存MappedByteBuffer mappedByteBuffer = new RandomAccessFile("input.txt", "rw").getChannel().map(FileChannel.MapMode.READ_WRITE, 0, 4);mappedByteBuffer.put(0, (byte) 'a');mappedByteBuffer.put(0, (byte) 'a');mappedByteBuffer.put(0, (byte) 'a');} }

NIO之網絡IO

使用NIO進行網絡非阻塞式編程, NIO編程模型圖:

Selector: 檢測Channel是否存在事件發生 ServerSocketChannel: 服務器 Channel: 管道 Client: 客戶端

NIO網絡編程結構圖:

selectionKey的4種狀態:OP_ACCEPT: 網絡已連接 value = 16OP_CONNECT: 連接已建立 value = 8OP_READ OP_WRITE: 讀/寫操作, value = 1 或value = 4

根據上面結構圖編寫簡單案例代碼:
NioServer

/*** server*/ public class NioServer {public static void main(String[] args) throws IOException, InterruptedException {ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.configureBlocking(false);serverSocketChannel.bind(new InetSocketAddress(9999));Selector selector = Selector.open();SelectionKey selectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);while (true) {//設置selector監控, 監控Channel的連接情況, 此處除去第一次的ServerSocketChannel注冊, 監控時長2sif (selector.select(2000) == 0) {System.out.println("------當前沒有要處理的Channel, 我去處理其他事------");TimeUnit.SECONDS.sleep(1);continue;}//此時有Channel進行連接, 獲取selectedKeys, 處理每一個Channel的對象事件Set<SelectionKey> selectionKeys = selector.selectedKeys();selectionKeys.forEach(key -> {if (key.isAcceptable()) {System.out.println("OP_ACCEPT");//說明此處的key對應的是最開始前面register的ServerSocketChannelServerSocketChannel server = (ServerSocketChannel) key.channel();//此處對象的hash值相同System.out.println("(ServerSocketChannel) key.channel(): " + server.hashCode());System.out.println("ServerSocketChannel.open(): " + serverSocketChannel.hashCode());try {SocketChannel socketChannel = serverSocketChannel.accept();socketChannel.configureBlocking(false);//此處注冊的時候就附加一個緩沖區, 可用于傳輸對象//當執行register的時候, 就會生成一個對應的SelectionKey事件, 當前的selectionKeys遍歷完成, 就會將該selectionKey添加到set集合中SelectionKey socketChannelReadRegister = socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));} catch (IOException e) {e.printStackTrace();}}if (key.isReadable()) {SocketChannel socketChannel = (SocketChannel) key.channel();ByteBuffer byteBuffer = (ByteBuffer) key.attachment();try {socketChannel.read(byteBuffer);System.out.println("客戶端內容: " + new String(byteBuffer.array(), StandardCharsets.UTF_8));} catch (IOException e) {e.printStackTrace();}}//這里使用remove與下面使用clear效果一樣, 都是清除當前已經執行過的Channel, 避免重復執行//當前的Channel對應的事件被處理過, 就不再被處理, 使用這種寫法較為恰當selectionKeys.remove(key);}); // selectionKeys.clear();}} }

NioClient

/*** client*/ public class NioClient {public static void main(String[] args) throws IOException, InterruptedException {SocketChannel socketChannel = SocketChannel.open();socketChannel.configureBlocking(false);if (!socketChannel.connect(new InetSocketAddress("127.0.0.1", 9999))) {//連接服務端失敗, 使用finishConnect進行連接, 此處finishConnect非阻塞while (!socketChannel.finishConnect()) {System.out.println("連接同時, 干其他事情");TimeUnit.SECONDS.sleep(2);}}//return new HeapByteBuffer(capacity, capacity); // ByteBuffer.allocate(1024);//return new HeapByteBuffer(array, offset, length);ByteBuffer buffer = ByteBuffer.wrap("hello world".getBytes(StandardCharsets.UTF_8));socketChannel.write(buffer);//此處阻塞, 若關閉連接, server會拋出異常System.out.println("進入睡眠");Thread.currentThread().join();} }

Netty初窺門徑

• Netty模型:
  

BossGroup: 處理客戶端連接請求 WorkGroup: 處理網絡讀寫操作 二者使用NioEventLoopGroup不斷循環處理任務線程, NioEventLoopGroup內部都有一個selector, 監聽每個Channel連接情況NioEventLoopGroup內部使用串行化設計: 消息讀取->解碼->處理->編碼->發送

• NioEventLoopGroup模型:
  

一個NioEventLoopGroup包含多個NioEventLoop 一個NioEventLoop包含一個Selector, 一個任務隊列 每個NioChannel都會綁定一個自己的ChannelPipeline

• ChannelPipeline模型:
  

ChannelPipeline: Handler集合, 負責處理/攔截inbound, outbound操作 ChannelHandlerContext: 事件處理器上下文對象, 內部包含每個具體的ChannelHandler, 也綁定對應Channel, pipeline信息

• 簡單入門案例:
  NettyServer:

/*** 服務器端*/ public class NettyServer {public static void main(String[] args) throws InterruptedException {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workGroup = new NioEventLoopGroup();//線程池中任務隊列數: ChannelOption.SO_BACKLOG, 128//讓連接保持活動狀態: ChannelOption.SO_KEEPALIVEServerBootstrap serverBootstrap = new ServerBootstrap().group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new NettyServerHandler());}});ChannelFuture channelFuture = serverBootstrap.bind(9999).sync();channelFuture.channel().closeFuture().sync();bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}

NettyServerHandler:

public class NettyServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf byteBuf = (ByteBuf) msg;System.out.println("客戶端發送的內容: " + byteBuf.toString(CharsetUtil.UTF_8));}/*** 數據讀取完成* @param ctx Channel上下文對象* @throws Exception*/@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("就是沒錢", CharsetUtil.UTF_8));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {//出現異常, 直接關閉上下文對象ctx.close();} }

NettyClient

/*** 客戶端*/ public class NettyClient {public static void main(String[] args) throws InterruptedException {EventLoopGroup workGroup = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap().group(workGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new NettyClientHandler());}});ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9999).sync();channelFuture.channel().closeFuture().sync();} }

NettyClientHandler:

public class NettyClientHandler extends ChannelInboundHandlerAdapter {/*** 通道準備就緒, 當前已經連接* @param ctx* @throws Exception*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("老板, 還錢吧".getBytes(CharsetUtil.UTF_8)));}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf byteBuf = (ByteBuf) msg;System.out.println(byteBuf.toString(CharsetUtil.UTF_8));} }

總結

以上是生活随笔為你收集整理的NIO之ByteBuffer_NIO之网络IO_与ChannelNetty初窥门径的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。