您的位置:首页 > 其它

软件开发中涉及操作系统知识总结—IO模型

2016-12-06 15:26 633 查看
有同步阻塞式IO,同步非阻塞IO,IO多路复用,异步IO等四种常见的IO模型。

一、理论:

1.同步阻塞式IO:

定义:用户线程在内核进行IO操作时被阻塞。

特点:简单、传统。

优劣:整个IO请求的过程中,用户线程是被阻塞的,这导致用户在发起IO请求时,不能做任何事情,对CPU的资源利用率不够。



2.同步非阻塞:

定义:在同步阻塞IO的基础上,将socket设置为NONBLOCK。发起IO请求时可立即返回。

特点:不断轮询。

优劣:耗费CPU资源。



3.IO复用:

定义:内核一旦发现进程指定的一个或者多个IO条件准备读取,它就通知该进程进行处理。

特点:阻塞,select ,poll epoll

优劣:同一个线程内同时处理多个IO请求。使用select函数的优点并不仅限于此。虽然上述方式允许单线程内处理多个IO请求,但是每个IO请求的过程还是阻塞的(在select函数上阻塞),平均时间甚至比同步阻塞IO模型还要长。如果用户线程只注册自己感兴趣的socket或者IO请求,然后去做自己的事情,等到数据到来时再进行处理,则可以提高CPU的利用率。



4.异步IO:

定义:“真正”的异步IO需要操作系统更强的支持。当用户线程收到通知时,数据已经被内核读取完毕,并放在了用户线程指定的缓冲区内,内核在IO完成后通知用户线程直接使用即可。

特点:异步,非阻塞,需要系统内核支持。

优劣:处理效率大大提升,cpu得到充分的利用。


                                                                                                                        

二、开发语言中的IO模型(java为例):

BIO-阻塞式IO:

在基于Socket的消息通信过程中,Socket服务端向外部提供服务,而Socket客户端可以建立到Socket服务端的连接,进而发送请求数据,然后等待Socket服务端处理,并返回处理结果(响应)。

package socket.simple;
import java.io.*;
import java.net.*;

/**
*  创建和使用ServerSocket的步骤:
1.    用一个您想让它侦听传入客户机连接的端口来实例化一个ServerSocket(如有问题则抛出 Exception)。
2.    调用ServerSocket的accept()以在等待连接期间造成阻塞。
3.    获取位于该底层Socket的流以进行读写操作。
4.    按使事情简单化的原则包装流。
5.    对Socket进行读写。
6.    关闭打开的流(并请记住,永远不要在关闭Writer之前关闭Reader)。
* @author hhh
*
*/
public class RemoteFileServer {
int listenPort;
public RemoteFileServer(int listenPort) {
this.listenPort=listenPort;
}
//允许客户机连接到服务器,等待客户机请求
public void acceptConnections() {
try {
ServerSocket server = new ServerSocket(listenPort);
Socket incomingConnection = null;
while(true) {
incomingConnection = server.accept();
handleConnection(incomingConnection);
}
}
catch(BindException e) {
System.out.println("Unable to bind to port "+listenPort);
}
catch(IOException e) {
System.out.println("Unable to instantiate a ServerSocket on port: "+listenPort);

}
}

//与客户机Socket交互以将客户机所请求的文件的内容发送到客户机
public void handleConnection(Socket incomingConnection) {
try {
OutputStream outputToSocket = incomingConnection.getOutputStream(); //
InputStream inputFromSocket = incomingConnection.getInputStream();  //
BufferedReader streamReader = new BufferedReader(new InputStreamReader(inputFromSocket));
FileReader fileReader = new FileReader(new File(streamReader.readLine()));
BufferedReader bufferedFileReader = new BufferedReader(fileReader);
PrintWriter streamWriter = new PrintWriter(outputToSocket);
String line = null;
while((line=bufferedFileReader.readLine())!=null){
streamWriter.println(line);
}
fileReader.close();
streamWriter.close();
streamReader.close();
}
catch(Exception e) {
System.out.println("Error handling a client: "+e);
e.printStackTrace();
}
}

public static void main(String args[]) {
RemoteFileServer server = new RemoteFileServer(1001);
server.acceptConnections();
}
}
package socket.simple;
import java.io.*;
import java.net.*;
/**
* 	创建和使用 Socket 的步骤:
1.    用您想连接的机器的 IP 地址和端口实例化 Socket(如有问题则抛出 Exception)。
2.    获取 Socket 上的流以进行读写。
3.    把流包装进 BufferedReader/PrintWriter 的实例,如果这样做能使事情更简单的话。
4.    对 Socket 进行读写。
5.    关闭打开的流。
* @author hhh
*
*/
public class RemoteFileClient {
protected BufferedReader socketReader;
protected PrintWriter socketWriter;
protected String hostIp;
protected int hostPort;

//构造方法
public RemoteFileClient(String hostIp, int hostPort) {
this.hostIp = hostIp;
this.hostPort=hostPort;
}

//向服务器请求文件的内容
public String getFile(String fileNameToGet) {
StringBuffer fileLines = new StringBuffer();
try {
socketWriter.println(fileNameToGet);
socketWriter.flush();
String line = null;
while((line=socketReader.readLine())!=null)
fileLines.append(line+"\n");
}
catch(IOException e) {
System.out.println("Error reading from file: "+fileNameToGet);
}
return fileLines.toString();
}

//连接到远程服务器
public void setUpConnection() {
try {
Socket client = new Socket(hostIp,hostPort);
socketReader = new BufferedReader(new InputStreamReader(client.getInputStream()));//得到输入流读取服务器发送的数据
socketWriter = new PrintWriter(client.getOutputStream());//得到输出流向服务器写数据
}
catch(UnknownHostException e) {
System.out.println("Error1 setting up socket connection: unknown host at "+hostIp+":"+hostPort);
}
catch(IOException e) {
System.out.println("Error2 setting up socket connection: "+e);
}
}

//断开远程服务器
public void tearDownConnection() {
try {
socketWriter.close(); //先关闭输出流
socketReader.close();//再关闭输入流
}catch(IOException e) {
System.out.println("Error tearing down socket connection: "+e);
}
}

public static void main(String args[]) {
RemoteFileClient remoteFileClient = new RemoteFileClient("127.0.0.1",1001);
remoteFileClient.setUpConnection();
StringBuffer fileContents = new StringBuffer();
fileContents.append(remoteFileClient.getFile("RemoteFileServer.java"));
//remoteFileClient.tearDownConnection();
System.out.println(fileContents);
}
}

NIO-非阻塞式IO:

JDK1.4提供了对非阻塞IO(NIO)的支持,JDK1.5_update10版本使用epoll替代了传统的select/poll,极大的提升了NIO通信的性能。

核心组成部分:channels 、buffers、selectors 。基本上,所有的 IO 在NIO 中都从一个Channel 开始。Channel 有点象流。 数据可以从Channel读到Buffer中,也可以从Buffer 写到Channel中。这里有个图示:



Selector
Selector允许单线程处理多个 Channel。如果你的应用打开了多个连接(通道),但每个连接的流量都很低,使用Selector就会很方便。例如,在一个聊天服务器中。
这是在一个单线程中使用一个Selector处理3个Channel的图示:



要使用Selector,得向Selector注册Channel,然后调用它的select()方法。这个方法会一直阻塞到某个注册的通道有事件就绪。一旦这个方法返回,线程就可以处理这些事件,事件的例子有如新连接进来,数据接收等。

nio服务端和客户端代码示例:

package cn.nio;

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;

/**
* NIO服务端
* @author 小路
*/
public class NIOServer {
//通道管理器
private Selector selector;

/**
* 获得一个ServerSocket通道,并对该通道做一些初始化的工作
* @param port  绑定的端口号
* @throws IOException
*/
public void initServer(int port) throws IOException {
// 获得一个ServerSocket通道
ServerSocketChannel serverChannel = ServerSocketChannel.open();
// 设置通道为非阻塞
serverChannel.configureBlocking(false);
// 将该通道对应的ServerSocket绑定到port端口
serverChannel.socket().bind(new InetSocketAddress(port));
// 获得一个通道管理器
this.selector = Selector.open();
//将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后,
//当该事件到达时,selector.select()会返回,如果该事件没到达selector.select()会一直阻塞。
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
}

/**
* 采用轮询的方式监听selector上是否有需要处理的事件,如果有,则进行处理
* @throws IOException
*/
@SuppressWarnings("unchecked")
public void listen() throws IOException {
System.out.println("服务端启动成功!");
// 轮询访问selector
while (true) {
//当注册的事件到达时,方法返回;否则,该方法会一直阻塞
selector.select();
// 获得selector中选中的项的迭代器,选中的项为注册的事件
Iterator ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = (SelectionKey) ite.next();
// 删除已选的key,以防重复处理
ite.remove();
// 客户端请求连接事件
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key
.channel();
// 获得和客户端连接的通道
SocketChannel channel = server.accept();
// 设置成非阻塞
channel.configureBlocking(false);

//在这里可以给客户端发送信息哦
channel.write(ByteBuffer.wrap(new String("向客户端发送了一条信息").getBytes()));
//在和客户端连接成功之后,为了可以接收到客户端的信息,需要给通道设置读的权限。
channel.register(this.selector, SelectionKey.OP_READ);

// 获得了可读的事件
} else if (key.isReadable()) {
read(key);
}

}

}
}
/**
* 处理读取客户端发来的信息 的事件
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException{
// 服务器可读取消息:得到事件发生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
// 创建读取的缓冲区
ByteBuffer buffer = ByteBuffer.allocate(10);
channel.read(buffer);
byte[] data = buffer.array();
String msg = new String(data).trim();
System.out.println("服务端收到信息:"+msg);
ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
channel.write(outBuffer);// 将消息回送给客户端
}

/**
* 启动服务端测试
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOServer server = new NIOServer();
server.initServer(8000);
server.listen();
}

}

package cn.nio;

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;

/**
* NIO客户端
* @author 小路
*/
public class NIOClient {
//通道管理器
private Selector selector;

/**
* 获得一个Socket通道,并对该通道做一些初始化的工作
* @param ip 连接的服务器的ip
* @param port  连接的服务器的端口号
* @throws IOException
*/
public void initClient(String ip,int port) throws IOException {
// 获得一个Socket通道
SocketChannel channel = SocketChannel.open();
// 设置通道为非阻塞
channel.configureBlocking(false);
// 获得一个通道管理器
this.selector = Selector.open();

// 客户端连接服务器,其实方法执行并没有实现连接,需要在listen()方法中调
//用channel.finishConnect();才能完成连接
channel.connect(new InetSocketAddress(ip,port));
//将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_CONNECT事件。
channel.register(selector, SelectionKey.OP_CONNECT);
}

/**
* 采用轮询的方式监听selector上是否有需要处理的事件,如果有,则进行处理
* @throws IOException
*/
@SuppressWarnings("unchecked")
public void listen() throws IOException {
// 轮询访问selector
while (true) {
selector.select();
// 获得selector中选中的项的迭代器
Iterator ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = (SelectionKey) ite.next();
// 删除已选的key,以防重复处理
ite.remove();
// 连接事件发生
if (key.isConnectable()) {
SocketChannel channel = (SocketChannel) key
.channel();
// 如果正在连接,则完成连接
if(channel.isConnectionPending()){
channel.finishConnect();

}
// 设置成非阻塞
channel.configureBlocking(false);

//在这里可以给服务端发送信息哦
channel.write(ByteBuffer.wrap(new String("向服务端发送了一条信息").getBytes()));
//在和服务端连接成功之后,为了可以接收到服务端的信息,需要给通道设置读的权限。
channel.register(this.selector, SelectionKey.OP_READ);

// 获得了可读的事件
} else if (key.isReadable()) {
read(key);
}

}

}
}
/**
* 处理读取服务端发来的信息 的事件
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException{
//和服务端的read方法一样
}

/**
* 启动客户端测试
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOClient client = new NIOClient();
client.initClient("localhost",8000);
client.listen();
}

}

NIO框架:Netty

Netty架构按照Reactor模式设计和实现,它的服务端通信序列图如下:



客户端通信序列图如下:



Netty的IO线程NioEventLoop由于聚合了多路复用器Selector,可以同时并发处理成百上千个客户端Channel,由于读写操作都是非阻塞的,这就可以充分提升IO线程的运行效率,避免由于频繁IO阻塞导致的线程挂起。另外,由于Netty采用了异步通信模式,一个IO线程可以并发处理N个客户端连接和读写操作,这从根本上解决了传统同步阻塞IO一连接一线程模型,架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

NIO框架MINA:

Apache Mina Server 是一个网络通信应用框架,也就是说,它主要是对基于TCP/IP、UDP/IP协议栈的通信框架(当然,也可以提供JAVA 对象的序列化服务、虚拟机管道通信服务等),Mina 可以帮助我们快速开发高性能、高扩展性的网络通信应用,Mina 提供了事件驱动、异步(Mina 的异步IO 默认使用的是JAVA NIO 作为底层支持)操作的编程模型。

Mina的基础应用分为三层:
· I/O Service -执行真实 I/O
· I/O Filter Chain -过滤或转化bytes为期望数据结构,正反互逆转换
· I/O Handler -真实的逻辑处理



服务端架构:

IOAcceptor 监听网络进来的连接或数据包

· 通过IP地址或端口为新连接或新session创建一个有序请求,并组合处理session
· IO Filters 所有数据包接收会话,遍历图中指定的过滤器链,过滤器可以用来修改数据包的内容(如对象转换、添加或移除信息等),从原始字节转换成/高级对象,PacketEncoder /Decoder尤其有用
· 最后包转换对象IOHandler. IOHandlers 能够用于业务需求



客户端架构:

客户端首先创建一个 IOConnector (MINA 构造Socket连接), 与服务器发起绑定访问
· 一旦连接创建,同时一个session也会被创建并关联到这个连接
· 应用或客户端写入session, 致使数据发送的服务器端,之后在过滤器链中遍历
· 所有从服务端接收到的相应或消息都是通过过滤器连对应的 IOHandler进行处理



3.AIO-异步IO:

JDK7对AIO的支持主要提供如下的一些封装类:

AsynchronousChannel:所有AIO Channel的父类。

AsynchronousByteChannel:支持Byte读写的Channel

AsynchronousDatagramChannel:支持数据包(datagram)读写的Channel

AsynchronousFileChannel:支持文件读写的Channel

AsynchronousServerSocketChannel:支持数据流读写的服务器端Channel

AsynchronousSocketChannel:支持数据流读写的客户端Channel

AsynchronousChannelGroup:支持资源共享的Channel分组 

在AIO socket编程中,服务端通道是AsynchronousServerSocketChannel,这个类提供了一个open()静态工厂,一个bind()方法用于绑定服务端IP地址(还有端口号),另外还提供了accept()用于接收用户连接请求。在客户端使用的通道是AsynchronousSocketChannel,这个通道处理提供open静态工厂方法外,还提供了read和write方法。

在AIO编程中,发出一个事件(accept read write等)之后要指定事件处理类(回调函数),AIO中的事件处理类是CompletionHandler<V,A>,这个接口定义了如下两个方法,分别在异步操作成功和失败时被回调。

 void completed(V result, A attachment);

 void failed(Throwable exc, A attachment);

AIO实现一个简单的TCP server:

package com.oks.test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class AIOEchoServer {

public final static int PORT = 8001;
public final static String IP = "127.0.0.1";

private AsynchronousServerSocketChannel server = null;

public AIOEchoServer(){
try {
//同样是利用工厂方法产生一个通道,异步通道 AsynchronousServerSocketChannel
server = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(IP,PORT));
} catch (IOException e) {
e.printStackTrace();
}
}

//使用这个通道(server)来进行客户端的接收和处理
public void start(){
System.out.println("Server listen on "+PORT);

//注册事件和事件完成后的处理器,这个CompletionHandler就是事件完成后的处理器
server.accept(null,new CompletionHandler<AsynchronousSocketChannel,Object>(){

final ByteBuffer buffer = ByteBuffer.allocate(1024);

@Override
public void completed(AsynchronousSocketChannel result,Object attachment) {

System.out.println(Thread.currentThread().getName());
Future<Integer> writeResult = null;

try{
buffer.clear();
result.read(buffer).get(100,TimeUnit.SECONDS);

System.out.println("In server: "+ new String(buffer.array()));

//将数据写回客户端
buffer.flip();
writeResult = result.write(buffer);
}catch(InterruptedException | ExecutionException | TimeoutException e){
e.printStackTrace();
}finally{
server.accept(null,this);
try {
writeResult.get();
result.close();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}

}

@Override
public void failed(Throwable exc, Object attachment) {
System.out.println("failed:"+exc);
}

});
}

public static void main(String[] args) {
new AIOEchoServer().start();
while(true){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

}
package com.oks.test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class AIOClient {

public static void main(String[] args) throws IOException {

final AsynchronousSocketChannel client = AsynchronousSocketChannel.open();

InetSocketAddress serverAddress = new InetSocketAddress("127.0.0.1",8001);

CompletionHandler<Void, ? super Object> handler = new CompletionHandler<Void,Object>(){

@Override
public void completed(Void result, Object attachment) {
client.write(ByteBuffer.wrap("Hello".getBytes()),null,
new CompletionHandler<Integer,Object>(){

@Override
public void completed(Integer result,
Object attachment) {
final ByteBuffer buffer = ByteBuffer.allocate(1024);
client.read(buffer,buffer,new CompletionHandler<Integer,ByteBuffer>(){

@Override
public void completed(Integer result,
ByteBuffer attachment) {
buffer.flip();
System.out.println(new String(buffer.array()));
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}

@Override
public void failed(Throwable exc,
ByteBuffer attachment) {
}

});
}

@Override
public void failed(Throwable exc, Object attachment) {
}

});
}

@Override
public void failed(Throwable exc, Object attachment) {
}

};

client.connect(serverAddress, null, handler);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

}
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: 
相关文章推荐