本文实例讲述了Python通过poll实现异步IO的方法。分享给大家供大家参考。具体分析如下:
在使用poll()后返回轮询对象,该对象支持以下方法:
pollObj.register(fd,[,eventmask])第一个参数是注册新的文件描述符fd,fd要么是一个整数文件描述符,要么可以带有一个获取文件描述符的fileno()方法的对象。eventmask是一些按位或标记,这些标记指示要处理的事件。
POLLIN: 用于读取数据
POLLPRI: 用于读取紧急数据
POLLOUT: 准备写入
POLLERR: 错误情况
POLLHUP: 保持状态
POLLNVAL: 无效请求
最后在循环中利用pollObj.poll()来进行对已注册的文件描述符进行轮询。返回一元祖(fd,event)。其中fd是文件描述符,event是指示时间的位掩码。至需要将event与对应的时间进行&测试即可。
利用poll创建对一个多路文件复制程序,代码如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
#!/usr/bin/env python import select BLKSIZE = 8192 def readwrite(fromfd,tofd): readbuf = fromfd.read(BLKSIZE) if readbuf: tofd.write(readbuf) tofd.flush() return len (readbuf) def copyPoll(fromfd1,tofd1,fromfd2,tofd2): #定义需要监听的事件 READ_ONLY = (select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR ) totalbytes = 0 if not (fromfd1 or fromfd2 or tofd1 or tofd2) : return 0 fd_dict = {fromfd1.fileno():fromfd1,fromfd2.fileno():fromfd2} #创建poll对象p p = select.poll() #利用poll对象p对需要监视的文件描述符进行注册 p.register(fromfd1,READ_ONLY) p.register(fromfd2,READ_ONLY) while True : #轮询已经注册的文件描述符是否已经准备好 result = p.poll() if len (result) ! = 0 : for fd,events in result: if fd_dict[fd] is fromfd1: if events & (select.POLLIN|select.POLLPRI): bytesread = readwrite(fromfd1,tofd1) totalbytes + = bytesread elif events & (select.POLLERR): p.unregister(fd_dict[fd]) if fd_dict[fd] is fromfd2: if events & (select.POLLIN|select.POLLPRI): bytesread = readwrite(fromfd2,tofd2) totalbytes + = bytesread elif events & (select.POLLERR): p.unregister(fd_dict[fd]) if bytesread < = 0 : break return totalbytes def main(): fromfd1 = open ( "/etc/fstab" , "r" ) fromfd2 = open ( "/root/VMwareTools-8.8.1-528969.tar.gz" , "r" ) tofd1 = open ( "/root/fstab" , "w+" ) tofd2 = open ( "/var/passwd" , "w+" ) totalbytes = copyPoll(fromfd1,tofd1,fromfd2,tofd2) print "Number of bytes copied %d\n" % totalbytes return 0 if __name__ = = "__main__" : main() |
希望本文所述对大家的Python程序设计有所帮助。