在使用poll()后返回轮询对象,该对象支持以下方法:
pollobj.register(fd,[,eventmask])第一个参数是注册新的文件描述符fd,fd要么是一个整数文件描述符,要么可以带有一个获取文件描述符的fileno()方法的对象。eventmask是一些按位或标记,这些标记指示要处理的事件。
pollin: 用于读取数据
pollpri: 用于读取紧急数据
pollout: 准备写入
pollerr: 错误情况
pollhup: 保持状态
pollnval: 无效请求
最后在循环中利用pollobj.poll()来进行对已注册的文件描述符进行轮询。返回一元祖(fd,event)。其中fd是文件描述符,event是指示时间的位掩码。至需要将event与对应的时间进行&测试即可。
利用poll创建对一个多路文件复制程序,代码如下:
#!/usr/bin/env pythonimport selectblksize=8192def 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 totalbytesdef 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 0if __name__==__main__: main()
希望本文所述对大家的python程序设计有所帮助。
