Python——IO多路复用之select模块poll方法

# settings.pyHOST = 'localhost'
PORT = 5555
buffersize = 1024
ADDR = HOST, PORT

# poll_server.pyfrom settings import *
from select import *
from socket import *s = socket()
s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) # 设置端口可立即重用
s.bind(ADDR)
s.listen()# 创建poll对象
p = poll()# 创建地图（字典）
fdmap = {s.fileno(): s}# 注册关注（事件）
p.register(s, POLLIN | POLLERR)
# POLLIN 对应select方法中的参数rlist -> 等待处理的IO事件
# POLLOUT 对应select方法中的参数wlist -> 主动处理的IO事件
# POLLERR 对应select方法中的参数xlist -> 出错处理的IO事件while True:
# 进行IO监控
# events = [(fileno, event), ...]
try:
events = p.poll()
except KeyboardInterrupt:
print('KeyboardInterrupt: Ctrl+C to exit')
break for fd, event in events:
if fd == s.fileno():
# 从地图中找到fd对应的对象
conn, addr = fdmap[fd].accept()
print('Connect from', addr)
# 注册关注
p.register(conn, POLLIN)
# 添加到地图中
fdmap[conn.fileno()] = conn
else:
data = fdmap[fd].recv(buffersize)
if not data:
p.unregister(fd) # 取消注册
fdmap[fd].close() # 关闭套接字
del fdmap[fd] # 从地图中删除
else:
print(fdmap[fd].getpeername(), data.decode())
fdmap[fd].send(b'Roger that!')
s.close()
print('El Fin')

# client.pyfrom socket import *
from settings import *s = socket()
s.connect(ADDR)while True:
data = input('>> ')
if not data:
break
s.send(data.encode())
print(s.recv(buffersize).decode())s.close()