python I/O多路复用包括3个模块，上一篇已经说过概念，这里我使用的是select模块实现一个ftp并发
服务器端核心代码：
1 import socket,select 2 import queue,os 3 from conf import setting 4 5 6 class select_server(object): 7 ''' ftp服务器核心程序''' 8 9 def __init__(self,ip,port): 10 self.server = socket.socket() 11 self.server.bind((ip,port)) 12 self.inputs = [self.server,] # 收 13 self.outputs = [] # 发 14 self.msg = {} # 保存客户端的socket信息及对应的登录账号，密码，登录状态和客户端消息队列 15 self.client_sock_cmd = {} # 以客户端socket为健名，值为对应客户端的所有信息 16 17 def listen(self): 18 ''' 启动监听并设定服务器为非阻塞模式''' 19 self.server.listen(100) 20 self.server.setblocking(False) 21 22 def login_auth(self,name,pwd,auth_file): 23 ''' 进行账户验证''' 24 if name in auth_file: 25 if pwd == auth_file[name]['passwd']: 26 return '200' 27 return '400' 28 return '400' 29 30 def sele(self): 31 ''' 非阻塞模式操作''' 32 while True: 33 # 监听客户端的连接，返回3个列表 34 readable, writeable, exceptionable = select.select(self.inputs, self.outputs, self.inputs) 35 # 客户端的连接 36 for s in readable: 37 if s is self.server: # 如果客户端为新连入的 38 conn,addr = self.server.accept() 39 print('已新建立一个客户端的连接',conn) 40 self.inputs.append(conn) # 添加客户端的socket到inputs列表 41 conn.setblocking(False) 42 self.msg[conn] = {} # 初始化客户端的值为一个队列 43 self.msg[conn]['queue'] = queue.Queue() 44 self.msg[conn]['account'] = [] 45 self.msg[conn]['status'] = False 46 conn.send('请输入登录账号:'.encode('utf-8')) 47 conn.send('请输入登录密码:'.encode('utf-8')) 48 else: 49 try: 50 data = s.recv(1024) 51 except ConnectionResetError as e: # 客户端中断捕捉的异常 52 exceptionable.append(s) 53 break 54 if data == b'exit': 55 exceptionable.append(s) 56 break 57 self.msg[s]['queue'].put(data) # 将客户端发来的数据保存到队列中 58 self.outputs.append(s) # 将客户端的socket添加到outputs列表 59 60 for w in writeable: # 客户端写列表 61 data = self.msg[w]['queue'].get() 62 # print(data) 63 # 判断账号是否登录，未登录执行如下 64 if self.msg[w]['status'] == False: 65 self.msg[w]['account'].append(data.decode()) 66 if len(self.msg[w]['account']) == 2: 67 # 进行账号密码验证操作 68 out = self.login_auth(self.msg[w]['account'][0],self.msg[w]['account'][1],setting.account) 69 if out == '200': 70 self.msg[w]['status'] = True 71 w.send('login'.encode('utf-8')) 72 else: 73 w.send('err'.encode('utf-8')) 74 exceptionable.append(w) 75 else: 76 # 已登录执行如下 77 # 根据客户端socket进行get和put操作 78 if w in self.client_sock_cmd: 79 if self.client_sock_cmd[w]['cmd'] == b'put': 80 # 进行put上传操作 81 self.client_sock_cmd[w]['file_io'].write(data) 82 self.client_sock_cmd[w]['file_io'].flush() 83 if os.path.getsize(self.client_sock_cmd[w]['filename']) == self.client_sock_cmd[w]['filesize']: 84 self.client_sock_cmd[w]['file_io'].close() 85 del self.client_sock_cmd[w] 86 print('文件保存完毕') 87 else: 88 # 进行get下载操作 89 if data.decode() == self.client_sock_cmd[w]['status']: 90 # 进行数据get操作在Linux必须要用send指定字节大小，否则报错，在windows下可以一次性发送sendall 91 # 由于这里使用了select多路复用模式，所以需要每次收发，都需要和客户端进行一次交互动作 92 if self.client_sock_cmd[w]['filesize'] - self.client_sock_cmd[w]['initsize'] >= 1024: 93 size = 1024 94 else: 95 size = self.client_sock_cmd[w]['filesize'] - self.client_sock_cmd[w]['initsize'] 96 data = self.client_sock_cmd[w]['file_io'].read(size) 97 self.client_sock_cmd[w]['initsize'] += len(data) 98 w.send(data) 99 if self.client_sock_cmd[w]['filesize'] == self.client_sock_cmd[w]['initsize']: 100 self.client_sock_cmd[w]['file_io'].close() 101 del self.client_sock_cmd[w] 102 print('文件发送完毕') 103 else: 104 # 根据客户端的socket进行字典初始化 105 self.client_sock_cmd[w] = {} 106 self.client_sock_cmd[w]['cmd'] = data.split()[0] 107 if self.client_sock_cmd[w]['cmd'] == b'put': 108 # 进行put的初始化操作 109 self.client_sock_cmd[w]['filename'], self.client_sock_cmd[w]['filesize'] = data.split()[1], int(data.split()[2]) 110 self.client_sock_cmd[w]['filename'] = os.path.join(setting.save_dir,self.client_sock_cmd[w]['filename']) 111 f = open(self.client_sock_cmd[w]['filename'], 'wb') 112 self.client_sock_cmd[w]['file_io'] = f 113 w.send('200'.encode('utf-8')) 114 else: 115 # 进行get的初始化操作 116 self.client_sock_cmd[w]['filename'] = data.split()[1] 117 self.client_sock_cmd[w]['filename'] = os.path.join(setting.save_dir,self.client_sock_cmd[w]['filename']) 118 if os.path.isfile(self.client_sock_cmd[w]['filename']): 119 self.client_sock_cmd[w]['status'] = '200' 120 self.client_sock_cmd[w]['initsize'] = 0 121 self.client_sock_cmd[w]['filesize'] = os.path.getsize(self.client_sock_cmd[w]['filename']) 122 status_msg = '%s %s' % (self.client_sock_cmd[w]['status'], str(self.client_sock_cmd[w]['filesize'])) 123 f = open(self.client_sock_cmd[w]['filename'], 'rb') 124 self.client_sock_cmd[w]['file_io'] = f 125 else: 126 status_msg = '%s 0' % ('404') 127 exceptionable.append(w) 128 w.send(status_msg.encode('utf-8')) 129 self.outputs.remove(w) 130 # 客户端异常或者退出清空对应的客户端socket信息并关闭连接 131 for e in exceptionable: 132 print('客户端退出:',e) 133 if e in self.msg: 134 del self.msg[e] 135 if e in self.outputs: 136 self.outputs.remove(e) 137 self.inputs.remove(e) 138 e.close()