nio涉及的類和方法
buffer
nio的buffer本質(zhì)上是，一塊內(nèi)存區(qū)域。被封裝到Buffer類里，并提供一組方法，供（channel）讀寫數(shù)據(jù)。
- 讀寫數(shù)據(jù)分如下4個(gè)步驟：
- 寫入數(shù)據(jù)到Buffer
- 調(diào)用flip()方法
- 從Buffer中讀取數(shù)據(jù)
- 調(diào)用clear()方法或者compact()方法
當(dāng)向buffer寫入數(shù)據(jù)時(shí)，buffer會(huì)記錄下寫了多少數(shù)據(jù)。一旦要讀取數(shù)據(jù)，需要通過(guò)flip()方法將Buffer從寫模式切換到讀模式。在讀模式下，可以讀取之前寫入到buffer的所有數(shù)據(jù)。
一旦讀完了所有的數(shù)據(jù)，就需要清空緩沖區(qū)，讓它可以再次被寫入。有兩種方式能清空緩沖區(qū)：調(diào)用clear()或compact()方法。clear()方法會(huì)清空整個(gè)緩沖區(qū)。compact()方法只會(huì)清除已經(jīng)讀過(guò)的數(shù)據(jù)。任何未讀的數(shù)據(jù)都被移到緩沖區(qū)的起始處，新寫入的數(shù)據(jù)將放到緩沖區(qū)未讀數(shù)據(jù)的后面
- buffer的分配
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
- buffer的類型
- ByteBuffer
- MappedByteBuffer
- CharBuffer
- DoubleBuffer
- FloatBuffer
- IntBuffer
- LongBuffer
- ShortBuffer
- buffer中寫數(shù)據(jù)
一種是調(diào)用buffer的put方法，一種是從channel中獲取。
byteBuffer.put(new byte[127]);
//channel read是buffer的寫
clientChannel.read(byteBuffer);
- buffer讀數(shù)據(jù)
讀數(shù)據(jù)前要調(diào)用flip()，切換到讀模式下。
一種是調(diào)用buffer的get方法，一種是往channel寫入
//保證讀取的內(nèi)容從頭開(kāi)始，并且讀取的長(zhǎng)度是已經(jīng)寫入的長(zhǎng)度
byteBuffer.flip();
byteBuffer.get();
clientChannel.write(byteBuffer);
- clear()與compact()方法
讀完Buffer中的數(shù)據(jù)，如果需要讓Buffer準(zhǔn)備好再次被寫入?？梢酝ㄟ^(guò)clear()或compact()方法來(lái)完成。
如果Buffer中有一些未讀的數(shù)據(jù)，調(diào)用clear()方法，數(shù)據(jù)將“被遺忘”，意味著不再有任何標(biāo)記會(huì)告訴你哪些數(shù)據(jù)被讀過(guò)，哪些還沒(méi)有。
如果Buffer中仍有未讀的數(shù)據(jù)，且后續(xù)還需要這些數(shù)據(jù)，但是此時(shí)想要先先寫些數(shù)據(jù)，那么使用compact()方法。
select
- select的創(chuàng)建
Selector selector = Selector.open();
- select()方法
select方法會(huì)阻塞
reactor模型在nio的基礎(chǔ)上
（1）有個(gè)主線程（Accept）去接收讀寫事件。具體的處理線程（Process）去處理讀寫請(qǐng)求（使用線程池）。
（2）引入隊(duì)列的概念，比如把寫請(qǐng)求事件放到隊(duì)列里，process去消費(fèi)隊(duì)列，去處理寫處理。*
還是以echo功能為例子。事件簡(jiǎn)單的reactor模型
package com.lxqn.jiapeng.reactorO;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;
/**
* 主線程 Accept功能 接收器
* Created by jiapeng on 2017/10/7.
*/
public class NIOServer {
public static void main(String[] args) throws IOException {
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(9998));
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
int coreNum = Runtime.getRuntime().availableProcessors();
Processor[] processors = new Processor[coreNum];
for (int i = 0; i < processors.length; i++) {
processors[i] = new Processor();
}
int index = 0;
while (true) {
int n=selector.select(1000);
if (n == 0) {
System.out.print(".");
continue;
}
System.out.println("n="+n);
if(n>0){
Set<SelectionKey> keys = selector.selectedKeys();
for (SelectionKey key : keys) {
keys.remove(key);
if (key.isAcceptable()) {
ServerSocketChannel acceptServerSocketChannel = (ServerSocketChannel) key.channel();
SocketChannel socketChannel = acceptServerSocketChannel.accept();
socketChannel.configureBlocking(false);
System.out.println("Accept request from {}" + socketChannel.getRemoteAddress());
Processor processor = processors[(int) ((index++) % coreNum)];
processor.addChannel(socketChannel);
processor.wakeup();
}
}
}
}
}
}
package com.lxqn.jiapeng.reactorO;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 具體處理網(wǎng)絡(luò)IO的類
* Created by jiapeng on 2017/10/7.
*/
public class Processor {
//使用線程池，大小為核數(shù)的2倍
private static final ExecutorService service =
Executors.newFixedThreadPool(2 * Runtime.getRuntime().availableProcessors());
private Selector selector;
//寫操作隊(duì)列，
protected ConcurrentLinkedQueue<SelectionKey> toWrite;
public Processor() throws IOException {
this.selector = SelectorProvider.provider().openSelector();
toWrite=new ConcurrentLinkedQueue<SelectionKey>();
start();
}
public void addChannel(SocketChannel socketChannel) throws ClosedChannelException {
socketChannel.register(this.selector, SelectionKey.OP_READ,ByteBuffer.allocate(1024));
}
public void wakeup() {
this.selector.wakeup();
}
public void start() {
service.submit(() -> {
while (true) {
int n=selector.select(1000);
if (n == 0) {
continue;
}
if(n>0){
System.out.println("process n="+n);
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isReadable()) {
ByteBuffer buffer = (ByteBuffer) key.attachment();
SocketChannel socketChannel = (SocketChannel) key.channel();
int count = socketChannel.read(buffer);
System.out.println("count="+count);
if (count < 0) {
socketChannel.close();
key.cancel();
System.out.println("Read ended"+socketChannel);
continue;
} else if (count == 0) {
System.out.println("Message size is 0"+socketChannel);
continue;
} else {
System.out.println("Read message"+socketChannel+" message="+new String(buffer.array()));
key.interestOps(SelectionKey.OP_WRITE);
}
}
if (key.isWritable()) {
System.out.println("start write");
toWrite.offer(key);
}
//模擬隊(duì)列處理的操作
SelectionKey writableKey=null;
while((writableKey=toWrite.poll())!=null) {
System.out.println("toWrite operation");
write(writableKey);
}
}
}
}
});
}
private void write(SelectionKey key) throws IOException{
ByteBuffer buf = (ByteBuffer) key.attachment();
//make buffer ready for read
buf.flip();
SocketChannel clientChannel = (SocketChannel) key.channel();
clientChannel.write(buf);
if (!buf.hasRemaining()) {
//交替注冊(cè)讀寫操作，實(shí)現(xiàn)echo的功能
key.interestOps(SelectionKey.OP_READ);
}
//清空buffer，清理已讀取長(zhǎng)度的buffer
buf.compact();
System.out.println("toWrite"+buf.toString()+" message="+new String(buf.array()));
}
}
