在学习NIO之前需要先整理清楚一些概念,以下都是我之前收集的一些笔记:
1.堵塞和非堵塞:堵塞和非堵塞是进程在访问数据的时候,数据内容是否准备就绪的一种处理方式。当数据没有准备时,堵塞:往往需要等待缓冲区的数据准备好后才处理其他的事情,否则一直在等待在那里。
非堵塞:当我们的进程访问我们的数据缓冲区的时候,数据没有准备好的时候,直接返回不需要等待。
2.同步和异步的方式
同步和异步都是基于应用程序和操作系统处理IO时间锁采用的方式。比如同步应用程序要直接参与IO读写的操作,异步:所有的IO读写交给操作系统去处理。同步的方式在处理IO事件的时候 必须阻止在某个方法上面等待我们的IO时间完成(堵塞IO事件或通过轮询IO事件的方式),对于异步来说,所有的IO读写都交给了操作系统。这个时候,我们可以去做其他的事情,并不需要去完成真正的IO操作,当操作完成IO后,给我们的应用程序一个通知就可以了。
同步:1.堵塞到IO事件 阻塞到read 或者 write ,这个时候我们就完全不能做自己的事件。让读写方法写入到线程里面,然后堵塞线程来实现,对线程的性能开销比较大
3 .IO事件的轮询 - 多路复用技术(select模式)
(扩展关于多路复用技术,有兴趣的可以先参考这篇文章,跟系统相关的,涉及到CPU的一些内核态和进程态等等,不一定要深入的去理解):
https://mp.weixin.qq.com/s?__biz=MzAxODI5ODMwOA==&mid=2666538919&idx=1&sn=6013c451b5f14bf809aec77dd5df6cff&scene=21#wechat_redirect
)
读写事件交给一个单独的线程来处理,这个完成IO事件的注册供,还有就是不断的去轮询我们的读写缓存区 看是否有数据准备好。通知我们通知相应的读写线程。这样的话,以前的读写线程就可以做其他的事件,这个时候堵塞的不是所有的IO线程 阻塞的select这个线程,
client select Boss
当客人来时,就给管家说 我来了 管家得到这个注册消息后,给boss说,我这里有一个或者多个客人,boss你去给某人A这件东西,给另外b这件东西,这个时候,客人可以做自己的事情的,比如看看花园等等,当管家知道boss给他任务后,他就会去找对应的某人,告诉他,boss给他某样东西。(根据我们的注册信息)。
4.缓冲区
Buffer是一个对象,包含一些要写入或者读出的数据。
缓冲区(Buffer)就是在内存中预留指定大小的存储空间用来对输入/输出(I/O)的数据作临时存储,这部分预留的内存空间就叫做缓冲区:
在NIO库中,所有数据都是用缓冲区处理的。在读取数据时,它是直接读到缓冲区中的;在写入数据时,也是写入到缓冲区中。任何时候访问NIO中的数据,都是通过缓冲区进行操作。
缓冲区实际上是一个数组,并提供了对数据结构化访问以及维护读写位置等信息。
具体的缓存区有这些:ByteBuffe、CharBuffer、 ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。他们实现了相同的接口:Buffer。
使用缓冲区有这么两个好处:
1、减少实际的物理读写次数
2、缓冲区在创建时就被分配内存,这块内存区域一直被重用,可以减少动态分配和回收内存的次数。
- Fields
所有缓冲区都有4个属性:capacity、limit、position、mark,并遵循:mark <= position <= limit <= capacity,下表格是对着4个属性的解释:
6.通道Channel(重点)
我们对数据的读取和写入要通过Channel,它就像水管一样,是一个通道。通道不同于流的地方就是通道是双向的,可以用于读、写和同时读写操作。
底层的操作系统的通道一般都是全双工的,所以全双工的Channel比流能更好的映射底层操作系统的API。
Channel主要分两大类:
SelectableChannel:用户网络读写
FileChannel:用于文件操作
后面代码会涉及的ServerSocketChannel和SocketChannel都是SelectableChannel的子类。
7.重点来了(多路复用器 Selector)
Selector是Java NIO 编程的基础。
Selector提供选择已经就绪的任务的能力:Selector会不断轮询注册在其上的Channel,如果某个Channel上面发生读或者写事件,这个Channel就处于就绪状态,会被Selector轮询出来,然后通过SelectionKey可以获取就绪Channel的集合,进行后续的I/O操作。
一个Selector可以同时轮询多个Channel,因为JDK使用了epoll()代替传统的select实现,所以没有最大连接句柄1024/2048的限制。所以,只需要一个线程负责Selector的轮询,就可以接入成千上万的客户端。
然后在深入的理解以上的一些概念后,你才能知道NIO的服务端与客户端之间的交互的每一步是什么原因?为什么要这样设置。
然后我们来对NIO的定义做一个说明
NIO我们一般认为是New I/O(也是官方的叫法),因为它是相对于老的I/O类库新增的(其实在JDK 1.4中就已经被引入了,但这个名词还会继续用很久,即使它们在现在看来已经是“旧”的了,所以也提示我们在命名时,需要好好考虑),做了很大的改变。但民间跟多人称之为Non-block I/O,即非阻塞I/O,因为这样叫,更能体现它的特点。而下文中的NIO,不是指整个新的I/O库,而是非阻塞I/O。
NIO提供了与传统BIO模型中的Socket和ServerSocket相对应的SocketChannel和ServerSocketChannel两种不同的套接字通道实现。
新增的两种通道都支持阻塞和非阻塞两种模式。
阻塞模式使用就像传统中的支持一样,比较简单,但是性能和可靠性都不好;非阻塞模式正好与之相反。
对于低负载、低并发的应用程序,可以使用同步阻塞I/O来提升开发速率和更好的维护性;对于高负载、高并发的(网络)应用,应使用NIO的非阻塞模式来开发。
对NIO与NBIO、AIO做一个对比:
BIO(jdk1.4之前,使用的都是BIO 就是阻塞IO)阻塞到我们的读写方法,阻塞到线程来提高性能,但是效果不好。对于线程的开销本来就是性能的浪费。
NIO(jdk1.4 之后 linux的多路复用技术 select模式)实现IO事件的轮询的方式 就是同步非堵塞的模式。本质是NIO 1.0对这种方式目前是主流的网络通信模式。Mina2.0 Netty5.0网络通信框架。通过这种实现的网络通信,比我们直接写的NIO要容易并且代码可读性会好
AIO(jdk 1.7过后 又叫NIO 2) 这才是实现真正的aio,也就是异步非阻塞模式,性能是最好的。底层实现是通过epoll的I/O多路复用机制。
建议要学习深入的话,一定要深入去研究I/O多路复用机制,就会明白为什么为什么能控制阻塞和是否异步?来提高性能。
可以参考我整理的这篇文章:
//TODO 待续上传
下面贴出代码,注意看上下文注释和上面的概念去理解哦。
服务器端:
Server
public class Server {
private static int DEFAULT_PORT = 12345;
private static ServerHandle serverHandle;
public static void start(){
start(DEFAULT_PORT);
}
public static synchronized void start(int port){
if(serverHandle!=null)
serverHandle.stop();
serverHandle = new ServerHandle(port);
new Thread(serverHandle,"Server").start();
}
public static void main(String[] args){
start();
}
}
执行的handle方法 ServerHandle
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import com.anxpp.io.utils.Calculator;
/**
* NIO服务端
* @author yangtao__anxpp.com
* @version 1.0
*/
public class ServerHandle implements Runnable{
private Selector selector;
private ServerSocketChannel serverChannel;
private volatile boolean started;
/**
* 构造方法
* @param port 指定要监听的端口号
*/
public ServerHandle(int port) {
try{
//创建选择器
selector = Selector.open();
//打开监听通道
serverChannel = ServerSocketChannel.open();
//如果为 true,则此通道将被置于阻塞模式;如果为 false,则此通道将被置于非阻塞模式
//与Selector一起使用时,Channel必须处于非阻塞模式下。这意味着不能将FileChannel与Selector一起使用,因为FileChannel不能切换到非阻塞模式。而套接字通道都可以。
serverChannel.configureBlocking(false);//开启非阻塞模式
//绑定端口 backlog设为1024
serverChannel.socket().bind(new InetSocketAddress(port),1024);
//监听客户端连接请求
//只要ServerSocketChannel及SocketChannel向Selector注册了特定的事件,Selector就会监控这些事件是否发生。
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
//标记服务器已开启
started = true;
System.out.println("服务器已启动,端口号:" + port);
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
}
public void stop(){
started = false;
}
@Override
public void run() {
//循环遍历selector
while(started){
try{
//无论是否有读写事件发生,selector每隔1s被唤醒一次
selector.select(1000);
//阻塞,只有当至少一个注册的事件发生的时候才会继续.
//selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key = null;
while(it.hasNext()){
key = it.next();
//每次迭代末尾的remove()调用,Selector不会自己从已选择集中移除SelectioKey实 例,必须在处理完通道时自己移除。
it.remove();
try{
handleInput(key);
}catch(Exception e){
if(key != null){
//会在再次获取已触犯事件的api时,进行cannel
key.cancel();
if(key.channel() != null){
key.channel().close();
}
}
}
}
}catch(Throwable t){
t.printStackTrace();
}
}
//selector关闭后会自动释放里面管理的资源
if(selector != null)
try{
selector.close();
}catch (Exception e) {
e.printStackTrace();
}
}
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
//处理新接入的请求消息
if(key.isAcceptable()){
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//通过ServerSocketChannel的accept创建SocketChannel实例
//完成该操作意味着完成TCP三次握手,TCP物理链路正式建立
SocketChannel sc = ssc.accept();
//设置为非阻塞的
sc.configureBlocking(false);
//注册为读
sc.register(selector, SelectionKey.OP_READ);
}
//读消息
if(key.isReadable()){
SocketChannel sc = (SocketChannel) key.channel();
//创建ByteBuffer,并开辟一个1M的缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
//读取请求码流,返回读取到的字节数
int readBytes = sc.read(buffer);
//读取到字节,对字节进行编解码
if(readBytes>0){
//将缓冲区当前的limit设置为position=0,用于后续对缓冲区的读取操作
buffer.flip();
//根据缓冲区可读字节数创建字节数组
byte[] bytes = new byte[buffer.remaining()];
//将缓冲区可读字节数组复制到新建的数组中
buffer.get(bytes);
String expression = new String(bytes,"UTF-8");
System.out.println("服务器收到消息:" + expression);
//处理数据
String result = null;
try{
result = Calculator.cal(expression).toString();
}catch(Exception e){
result = "计算错误:" + e.getMessage();
}
//发送应答消息
doWrite(sc,result);
}
//没有读取到字节 忽略
// else if(readBytes==0);
//链路已经关闭,释放资源
else if(readBytes<0){
key.cancel();
sc.close();
}
}
}
}
//异步发送应答消息
private void doWrite(SocketChannel channel,String response) throws IOException{
//将消息编码为字节数组
byte[] bytes = response.getBytes();
//根据数组容量创建ByteBuffer
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
//将字节数组复制到缓冲区
writeBuffer.put(bytes);
//flip操作
writeBuffer.flip();
//发送缓冲区的字节数组
channel.write(writeBuffer);
//****此处不含处理“写半包”的代码
}
}
可以看到,创建NIO服务端的主要步骤如下:
打开ServerSocketChannel,监听客户端连接
绑定监听端口,设置连接为非阻塞模式
创建Reactor线程,创建多路复用器并启动线程
将ServerSocketChannel注册到Reactor线程中的Selector上,监听ACCEPT事件
Selector轮询准备就绪的key
Selector监听到新的客户端接入,处理新的接入请求,完成TCP三次握手,简历物理链路
设置客户端链路为非阻塞模式
将新接入的客户端连接注册到Reactor线程的Selector上,监听读操作,读取客户端发送的网络消息
异步读取客户端消息到缓冲区
对Buffer编解码,处理半包消息,将解码成功的消息封装成Task
-
将应答消息编码为Buffer,调用SocketChannel的write将消息异步发送给客户端.
因为应答消息的发送,SocketChannel也是异步非阻塞的,所以不能保证一次能吧需要发送的数据发送完,此时就会出现写半包的问题。我们需要注册写操作,不断轮询Selector将没有发送完的消息发送完毕,然后通过Buffer的hasRemain()方法判断消息是否发送完成
NIO客户端
public class Client {
private static String DEFAULT_HOST = "127.0.0.1";
private static int DEFAULT_PORT = 12345;
private static ClientHandle clientHandle;
public static void start(){
start(DEFAULT_HOST,DEFAULT_PORT);
}
public static synchronized void start(String ip,int port){
if(clientHandle!=null)
clientHandle.stop();
clientHandle = new ClientHandle(ip,port);
new Thread(clientHandle,"Server").start();
}
//向服务器发送消息
public static boolean sendMsg(String msg) throws Exception{
if(msg.equals("q")) return false;
clientHandle.sendMsg(msg);
return true;
}
public static void main(String[] args){
start();
}
}
// ClientHandle:
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/**
* NIO客户端
* @author yangtao__anxpp.com
* @version 1.0
*/
public class ClientHandle implements Runnable{
private String host;
private int port;
private Selector selector;
private SocketChannel socketChannel;
private volatile boolean started;
public ClientHandle(String ip,int port) {
this.host = ip;
this.port = port;
try{
//创建选择器
selector = Selector.open();
//打开监听通道
socketChannel = SocketChannel.open();
//如果为 true,则此通道将被置于阻塞模式;如果为 false,则此通道将被置于非阻塞模式
socketChannel.configureBlocking(false);//开启非阻塞模式
started = true;
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
}
public void stop(){
started = false;
}
@Override
public void run() {
try{
doConnect();
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
//循环遍历selector
while(started){
try{
//无论是否有读写事件发生,selector每隔1s被唤醒一次
selector.select(1000);
//阻塞,只有当至少一个注册的事件发生的时候才会继续.
// selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.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);
}
}
//selector关闭后会自动释放里面管理的资源
if(selector != null)
try{
selector.close();
}catch (Exception e) {
e.printStackTrace();
}
}
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
SocketChannel sc = (SocketChannel) key.channel();
if(key.isConnectable()){
if(sc.finishConnect());
else System.exit(1);
}
//读消息
if(key.isReadable()){
//创建ByteBuffer,并开辟一个1M的缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
//读取请求码流,返回读取到的字节数
int readBytes = sc.read(buffer);
//读取到字节,对字节进行编解码
if(readBytes>0){
//将缓冲区当前的limit设置为position=0,用于后续对缓冲区的读取操作
buffer.flip();
//根据缓冲区可读字节数创建字节数组
byte[] bytes = new byte[buffer.remaining()];
//将缓冲区可读字节数组复制到新建的数组中
buffer.get(bytes);
String result = new String(bytes,"UTF-8");
System.out.println("客户端收到消息:" + result);
}
//没有读取到字节 忽略
// else if(readBytes==0);
//链路已经关闭,释放资源
else if(readBytes<0){
key.cancel();
sc.close();
}
}
}
}
//异步发送消息
private void doWrite(SocketChannel channel,String request) throws IOException{
//将消息编码为字节数组
byte[] bytes = request.getBytes();
//根据数组容量创建ByteBuffer
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
//将字节数组复制到缓冲区
writeBuffer.put(bytes);
//flip操作
writeBuffer.flip();
//发送缓冲区的字节数组
channel.write(writeBuffer);
//****此处不含处理“写半包”的代码
}
private void doConnect() throws IOException{
if(socketChannel.connect(new InetSocketAddress(host,port)));
else socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
public void sendMsg(String msg) throws Exception{
socketChannel.register(selector, SelectionKey.OP_READ);
doWrite(socketChannel, msg);
}
}
演示结果
首先运行服务器,顺便也运行一个客户端:
import java.util.Scanner;
/**
* 测试方法
* @author yangtao__anxpp.com
* @version 1.0
*/
public class Test {
//测试主方法
@SuppressWarnings("resource")
public static void main(String[] args) throws Exception{
//运行服务器
Server.start();
//避免客户端先于服务器启动前执行代码
Thread.sleep(100);
//运行客户端
Client.start();
while(Client.sendMsg(new Scanner(System.in).nextLine()));
}
}
1. 服务器已启动,端口号:12345`
2. `1+2+3+4+5+6`
3. `服务器收到消息:1+2+3+4+5+6`
4. `客户端收到消息:21`
5. `1*2/3-4+5*6/7-8`
6. `服务器收到消息:1*2/3-4+5*6/7-8`
7. `客户端收到消息:-7.0476190476190474`