nio涉及的类和方法
buffer
nio的buffer本质上是，一块内存区域。被封装到Buffer类里，并提供一组方法，供（channel）读写数据。
- 读写数据分如下4个步骤：
- 写入数据到Buffer
- 调用flip()方法
- 从Buffer中读取数据
- 调用clear()方法或者compact()方法
当向buffer写入数据时，buffer会记录下写了多少数据。一旦要读取数据，需要通过flip()方法将Buffer从写模式切换到读模式。在读模式下，可以读取之前写入到buffer的所有数据。
一旦读完了所有的数据，就需要清空缓冲区，让它可以再次被写入。有两种方式能清空缓冲区：调用clear()或compact()方法。clear()方法会清空整个缓冲区。compact()方法只会清除已经读过的数据。任何未读的数据都被移到缓冲区的起始处，新写入的数据将放到缓冲区未读数据的后面
- buffer的分配
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
- buffer的类型
- ByteBuffer
- MappedByteBuffer
- CharBuffer
- DoubleBuffer
- FloatBuffer
- IntBuffer
- LongBuffer
- ShortBuffer
- buffer中写数据
一种是调用buffer的put方法，一种是从channel中获取。
byteBuffer.put(new byte[127]);
//channel read是buffer的写
clientChannel.read(byteBuffer);
- buffer读数据
读数据前要调用flip()，切换到读模式下。
一种是调用buffer的get方法，一种是往channel写入
//保证读取的内容从头开始，并且读取的长度是已经写入的长度
byteBuffer.flip();
byteBuffer.get();
clientChannel.write(byteBuffer);
- clear()与compact()方法
读完Buffer中的数据，如果需要让Buffer准备好再次被写入。可以通过clear()或compact()方法来完成。
如果Buffer中有一些未读的数据，调用clear()方法，数据将“被遗忘”，意味着不再有任何标记会告诉你哪些数据被读过，哪些还没有。
如果Buffer中仍有未读的数据，且后续还需要这些数据，但是此时想要先先写些数据，那么使用compact()方法。
select
- select的创建
Selector selector = Selector.open();
- select()方法
select方法会阻塞
reactor模型在nio的基础上
（1）有个主线程（Accept）去接收读写事件。具体的处理线程（Process）去处理读写请求（使用线程池）。
（2）引入队列的概念，比如把写请求事件放到队列里，process去消费队列，去处理写处理。*
还是以echo功能为例子。事件简单的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;
/**
* 具体处理网络IO的类
* Created by jiapeng on 2017/10/7.
*/
public class Processor {
//使用线程池，大小为核数的2倍
private static final ExecutorService service =
Executors.newFixedThreadPool(2 * Runtime.getRuntime().availableProcessors());
private Selector selector;
//写操作队列，
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);
}
//模拟队列处理的操作
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()) {
//交替注册读写操作，实现echo的功能
key.interestOps(SelectionKey.OP_READ);
}
//清空buffer，清理已读取长度的buffer
buf.compact();
System.out.println("toWrite"+buf.toString()+" message="+new String(buf.array()));
}
}
网友评论