`
haiyupeter
  • 浏览: 427486 次
  • 性别: Icon_minigender_1
  • 来自: 广州
社区版块
存档分类
最新评论

异步I/O处理

 
阅读更多
异步I/O处理过程

异步I/O处理的优点
    I/O密集型计算(进程所执行的I/O操作比执行的处理操作更多)的任务中,使用异步I/O的方式,可以提高CPU对应用程序处理的吞吐率,应用程序无需进行I/O阻塞。保证在I/O处理时,仍能进行应用程序的处理
异步I/O的缺点
    如果程序的I/O操作频繁并且短暂,则I/O线程切换代价较大,并且异步I/O库处理有一定的开销(磁盘磁头的寻址切换等),异步I/O对程序的写法有一定的门槛。

异步I/O的应用场景
    NodeJS的单进程,单线程模型。为了能同时处理更多的请求,I/O操作,必需使用异步处理的方式,NodeJS采用的是libeio的实现方式。

异步I/O的实现
glibc:
在文件目录下新建file.txt文件,file.txt内容:hello world!

#include <aio.h>  
#include <errno.h>  
#include <stdio.h>  
#include <stdlib.h>  
#include <string.h>  
char bufferAO[8192] __attribute__((aligned(4096)));  
  
void aio_completion_handler(int signo, siginfo_t *info, void *context)  
{  
    int ret;  
        write(1, "callback\n", 9);  
    struct aiocb *req;  
    // Ensure it's our signal  
    if (info->si_signo == SIGIO) {  
      
        req = (struct aiocb *)info->si_value.sival_ptr;  
        printf("data: %s\n" ,req->aio_buf);  
        // Did the request complete?  
        if (aio_error( req ) == 0) {  
            // Request completed successfully, get the return status  
            ret = aio_return( req );  
        }  
  
    }  
    return;  
}  
  
int  main()  
{  
    int fd, ret;  
    struct sigaction sig_act;  
    struct aiocb my_aiocb;  
    fd = open("file.txt", O_RDONLY);  
    if(fd < 0){  
        perror("open fail");  
    }  
  
    // Set up the signal handler  
    sigemptyset(&sig_act.sa_mask);  
    sig_act.sa_flags = SA_SIGINFO;  
    sig_act.sa_sigaction = aio_completion_handler;  
  
  
    // Set up the AIO request  
    bzero( (char *)&my_aiocb, sizeof(struct aiocb) );  
    my_aiocb.aio_fildes = fd;  
    my_aiocb.aio_buf = malloc(sizeof(bufferAO)+1);  
    my_aiocb.aio_nbytes = sizeof(bufferAO);  
    my_aiocb.aio_offset = 0;  
  
      
    // Link the AIO request with the Signal Handler  
    my_aiocb.aio_sigevent.sigev_notify = SIGEV_SIGNAL;  
    my_aiocb.aio_sigevent.sigev_signo = SIGIO;  
    my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;  
  
    // Map the Signal to the Signal Handler  
    ret = sigaction( SIGIO, &sig_act, NULL );  
      
    ret = aio_read(&my_aiocb);  
    if (ret < 0) {  
        perror("error info");  
    }  
    write(1, "caller thread\n", 14);  
    sleep(5);  
}  

compile:  gcc aio_demo.c -o aio_demo.o -lrt
run: ./aio_demo.o
result:
caller thread
callback
data: hello world!
kernel native aio:
待完善

libeio:
libeio的源码下载:https://github.com/scunningham/libeio

libeio的编译及运行过程:

./autogen.sh

./configure

make

sudo make install

export LD_LIBRARY_PATH=".libs;$LD_LIBRARY_PATH"

#此时可以编译运行demo.c了

gcc demo.c -o demo -leio

./demo

运行结果:

具体的libeio源码及流程分析可以查看大牛已经做的文档:
NodeJS代码阅读笔记之libeio
libeio异步I/O初探

Java:
Server端:

package org.operamasks.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.nio.charset.Charset;
import java.util.Calendar;
import java.util.Iterator;


public class TestServer {
	public static void main(String[] args) {
		new Thread(new EchoServer(1982)).start();
	}
}

class EchoServer implements Runnable {
	//要监听的端口号
	private int port;
	//生成一个信号监视器
	private Selector s;
	//读缓冲区
	private ByteBuffer r_bBuf = ByteBuffer.allocate(1024);
	private ByteBuffer w_bBuf;
	
	public EchoServer(int port) {
		this.port = port;
		try {
			s = Selector.open();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void run() {
		try {
			//生成一个ServerScoket通道的实例对象,用于侦听可能发生的IO事件
			ServerSocketChannel ssc = ServerSocketChannel.open();
			//将该通道设置为异步方式
			ssc.configureBlocking(false);
			//绑定到一个指定的端口
			ssc.socket().bind(new InetSocketAddress(port));
			//注册特定类型的事件到信号监视器上
			ssc.register(s, SelectionKey.OP_ACCEPT);
			System.out.println("The server has been launched...");
			while(true) {
				//将会阻塞执行,直到有事件发生
				s.select();
				Iterator it = s.selectedKeys().iterator();
				while(it.hasNext()) {
					SelectionKey key = it.next();
					//key定义了四种不同形式的操作
					switch(key.readyOps()) {
					case SelectionKey.OP_ACCEPT :
						dealwithAccept(key);
						break;
					case SelectionKey.OP_CONNECT :
						break;
					case SelectionKey.OP_READ :
						dealwithRead(key);
						break;
					case SelectionKey.OP_WRITE :
						break;
					}
					//处理结束后移除当前事件,以免重复处理
					it.remove();
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	//处理接收连接的事件
	private void dealwithAccept(SelectionKey key) {
		try {
			System.out.println("新的客户端请求连接...");
			ServerSocketChannel server = (ServerSocketChannel)key.channel();
			SocketChannel sc = server.accept();
			sc.configureBlocking(false);
			//注册读事件
			sc.register(s, SelectionKey.OP_READ);
			System.out.println("客户端连接成功...");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	//处理客户端发来的消息,处理读事件
	private void dealwithRead(SelectionKey key) {
		try {
			SocketChannel sc = (SocketChannel)key.channel();
			System.out.println("读入数据");
			r_bBuf.clear();
			//将字节序列从此通道中读入给定的缓冲区r_bBuf
			sc.read(r_bBuf);
			r_bBuf.flip();
			String msg = Charset.forName("UTF-8").decode(r_bBuf).toString();
			if(msg.equalsIgnoreCase("time")) {
				w_bBuf = ByteBuffer.wrap(getCurrentTime().getBytes("UTF-8"));
				sc.write(w_bBuf);
				w_bBuf.clear();
			} else if(msg.equalsIgnoreCase("bye")) {
				sc.write(ByteBuffer.wrap("已经与服务器断开连接".getBytes("UTF-8")));
				sc.socket().close();
			} else {
				sc.write(ByteBuffer.wrap(msg.getBytes("UTF-8")));
			}
			System.out.println(msg);
			System.out.println("处理完毕...");
			r_bBuf.clear();
			try {
				Thread.currentThread();
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	private String getCurrentTime() {
		Calendar date = Calendar.getInstance();
		String time = "服务器当前时间:" +
					  date.get(Calendar.YEAR) + "-" +
					  date.get(Calendar.MONTH)+1 + "-" +
					  date.get(Calendar.DATE) + " " +
					  date.get(Calendar.HOUR) + ":" +
					  date.get(Calendar.MINUTE) + ":" +
					  date.get(Calendar.SECOND);
		return time;
	}
}


Client:

package org.operamasks.nio;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class TestClient {
	public static void main(String[] args) {
		new MiniClient("localhost", 1982);
	}
}
class MiniClient {
	private SocketChannel sc;
	private ByteBuffer w_bBuf;
	private ByteBuffer r_bBuf = ByteBuffer.allocate(1024);
	public MiniClient(String host, int port) {
		try {
			InetSocketAddress remote = new InetSocketAddress(host, port);
			sc = SocketChannel.open();
			sc.connect(remote);
			if(sc.finishConnect()) {
				System.out.println("已经与服务器成功建立连接...");
			}
			while(true) {
				if(!sc.socket().isConnected()) {
					System.out.println("已经与服务器失去了连接...");
					return ;
				}
				BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
				String str = br.readLine();
				System.out.println("读入一行数据,开始发送...");
				w_bBuf = ByteBuffer.wrap(str.getBytes("UTF-8"));
				//将缓冲区中数据写入通道
				sc.write(w_bBuf);
				System.out.println("数据发送成功...");
				w_bBuf.clear();
				System.out.println("接收服务器端响应消息...");
				try {
					Thread.currentThread();
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				r_bBuf.clear();
				//将字节序列从此通道中读入给定的缓冲区r_bBuf
				sc.read(r_bBuf);
				r_bBuf.flip();
				String msg = Charset.forName("UTF-8").decode(r_bBuf).toString();
				System.out.println(msg);				
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}



参考文章:

1.使用异步I/O大大提高应用程序的性能

2.Linux AIO(异步IO)那些事儿

3.NodeJS代码阅读笔记之libeio
  • 大小: 16.4 KB
  • 大小: 24.3 KB
分享到:
评论

相关推荐

    异步通知和异步I/O

    在Linux系统中,异步通知和异步I/O是设备驱动开发中的重要概念,它们极大地提高了系统的效率和响应性。本文将深入探讨这两个主题,并结合《Linux设备驱动开发详解》一书中的理论与实践,为你提供详尽的理解。 首先...

    MFC IOCP模型异步IO

    IOCP(I/O Completion Port,I/O完成端口)是Windows操作系统提供的一种高效的异步I/O模型,尤其适用于处理大量并发I/O操作。在这个场景中,"MFC IOCP模型异步IO"指的是使用MFC来实现基于IOCP的异步I/O操作。 IOCP...

    Performing I/O-Bound Asynchronous Operations

    借助.NET框架提供的丰富API,开发者能够轻松实现高效可靠的异步I/O处理机制。在未来的发展中,随着更多新技术的出现,异步编程将会变得更加便捷和强大。 --- 本文档总结了Jeffrey Richter关于I/O密集型异步操作的...

    netty高性能异步I/O服务器/客户端开源开发工具

    1. **异步I/O**:Netty采用非阻塞I/O模型,允许在等待I/O操作完成时执行其他任务,从而提高了系统的并发处理能力。在传统的阻塞I/O模型中,一个线程必须等待I/O操作完成才能继续执行,而Netty通过事件循环(Event...

    LINUX2.5下的异步I_O在FTP服务器上的应用.pdf

    在Linux操作系统中,异步I/O(Asynchronous Input/Output, AIO)是一种优化I/O性能的技术,尤其是在处理大量并发I/O操作的场景,如FTP(File Transfer Protocol)服务器。在Linux 2.5版本及其后续版本中,异步I/O已...

    windows下六种socket I/O模型示例

    - 对于异步I/O和IOCP,可能包含如何创建完成端口,以及如何处理I/O完成通知的示例代码。 - 每个示例可能包括详细的注释,解释了代码的运行流程和关键步骤。 通过学习和理解这些示例,你可以更好地掌握在Windows环境...

    一个异步 I/O 下载工具,异步下载器_python_代码_下载

    在IT领域,异步I/O(Asynchronous Input/Output)是一种高效的...在解压文件`aiodl-master`后,用户可以查看源代码,学习如何在Python中实现这样的异步下载器,这对于深入理解Python异步编程和网络I/O处理非常有帮助。

    GLIBC中Linux异步I_O函数的实现.pdf

    异步I/O的基本思想是,应用程序发起I/O请求后,不再等待操作完成,而是立即返回,由内核在后台处理I/O操作。当操作完成后,内核通过某种机制通知应用程序。这种方式提高了系统的并发性和效率,尤其是在处理大量并发I...

    nodejs异步IO的实现

    在IT行业中,Node.js以其高效的异步I/O处理能力而闻名。Node.js是建立在Google Chrome V8引擎上的JavaScript运行环境,它允许开发者使用JavaScript进行服务器端编程。本篇文章将深入探讨Node.js如何实现异步I/O,...

    高性能异步I_O适配层的设计与实现

    高并发负载下的Internet服务程序需要高效的I/O处理能力。同步I/O方式由于其阻塞性质,在面对大量并发请求时会显著降低性能。相比之下,异步I/O方式能够在设备进行I/O操作的同时让进程继续执行其他任务,因此具有更高...

    Socket I/O 模型的使用示例

    - 重叠I/O是Windows特有的另一种异步I/O模型,使用`WSASend`和`WSARecv`函数,允许在I/O操作进行的同时执行其他计算任务。 - 通过配合I/O完成端口(IOCP)使用,可以实现高度并发的网络服务,特别适合服务器端开发...

    C++/MFC实现winsock io异步操作

    - **原理**:重叠I/O是Windows系统提供的高级异步I/O模型,它允许I/O操作在发出后立即返回,而无需等待操作完成。 - **使用结构**:使用`OVERLAPPED`结构体记录I/O操作的状态,并通过`WSASend`和`WSARecv`等函数...

    重叠I/O类 代码

    4. **错误处理**:由于异步I/O可能导致错误在未来的某个时间点返回,所以类中需要有处理这些错误的机制,如检查`WSAGetLastError`返回值,并对可能出现的错误进行适当的处理。 5. **资源管理**:类应负责管理套接字...

    Java I/O, 2nd Edition

    3. **非阻塞I/O(NIO)**:Java NIO在Java 1.4引入,提供了异步I/O操作,可以处理多个连接,而无需为每个连接创建一个新的线程。关键类有Selector、Channel和Buffer,Selector可以监控多个通道的事件,Channel代表I/...

    1_WINSOCK的I/O模型_

    异步I/O可以最大限度地提高服务器的并发处理能力,适合处理大规模的群聊服务。 在实现C/S模式的群聊服务软件时,通常会选择I/O多路复用或异步I/O模型,因为它们能有效管理大量并发连接,确保服务器的高效运行。...

    unix平台下I/O聚集和分离的一种方案

    Unix系统提供了多种I/O模型,如阻塞I/O、非阻塞I/O、I/O多路复用(如select、poll、epoll)、信号驱动I/O以及异步I/O。I/O聚集通常指的是在一个系统调用中处理多个文件描述符,而I/O分离则是在不同时间或通过不同...

    nodejs异步IO的实现 转:http://cnodejs.org/topic/4f16442ccae1f4aa2700113b

    在IT领域,Node.js以其高效的异步I/O处理能力而著名。Node.js是基于Chrome V8引擎的JavaScript运行环境,它的出现使得JavaScript得以在服务器端运行,为开发高性能的网络应用提供了可能。本文将深入探讨Node.js异步...

    I/O完成端口例子

    通过这个示例,你可以学习如何创建和使用I/O完成端口,以及如何设计一个高效的并发I/O处理系统。完成端口的使用可以帮助你构建能够处理大规模并发请求的高性能应用程序,提升服务的稳定性和响应速度。在实践中不断...

    详解socket阻塞与非阻塞,同步与异步、I/O模型

    而Node.js推崇的是非阻塞I/O模型,通过事件驱动和回调函数实现异步操作,以提高系统的并发能力。 总结来说,同步与异步是关于处理结果获取方式的不同策略,而阻塞与非阻塞是关于处理过程中线程状态管理的差异。在...

Global site tag (gtag.js) - Google Analytics