python学习之-- IO多路复用 select模块

python I/O多路复用包括3个模块，上一篇已经说过概念，这里我使用的是select模块实现一个ftp并发

服务器端核心代码：

import socket,select
import queue,os
from conf import setting


class select_server(object):
    ''' ftp服务器核心程序'''

    def __init__(self,ip,port):
        self.server = socket.socket()
        self.server.bind((ip,port))
        self.inputs = [self.server,]  # 收
        self.outputs = []  # 发
        self.msg = {}  # 保存客户端的socket信息及对应的登录账号，密码，登录状态和客户端消息队列
        self.client_sock_cmd = {}  # 以客户端socket为健名，值为对应客户端的所有信息

    def listen(self):
        ''' 启动监听并设定服务器为非阻塞模式'''
        self.server.listen(100)
        self.server.setblocking(False)

    def login_auth(self,name,pwd,auth_file):
        ''' 进行账户验证'''
        if name in auth_file:
            if pwd == auth_file[name]['passwd']:
                return '200'
            return '400'
        return '400'

    def sele(self):
        ''' 非阻塞模式操作'''
        while True:
            # 监听客户端的连接，返回3个列表
            readable, writeable, exceptionable = select.select(self.inputs, self.outputs, self.inputs)
            # 客户端的连接
            for s in readable:
                if s is self.server:  # 如果客户端为新连入的
                    conn,addr = self.server.accept()
                    print('已新建立一个客户端的连接',conn)
                    self.inputs.append(conn)  # 添加客户端的socket到inputs列表
                    conn.setblocking(False)
                    self.msg[conn] = {}  # 初始化客户端的值为一个队列
                    self.msg[conn]['queue'] = queue.Queue()
                    self.msg[conn]['account'] = []
                    self.msg[conn]['status'] = False
                    conn.send('请输入登录账号:'.encode('utf-8'))
                    conn.send('请输入登录密码:'.encode('utf-8'))
                else:
                    try:
                        data = s.recv(1024)
                    except ConnectionResetError as e:  # 客户端中断捕捉的异常
                        exceptionable.append(s)
                        break
                    if data == b'exit':
                        exceptionable.append(s)
                        break
                    self.msg[s]['queue'].put(data)  # 将客户端发来的数据保存到队列中
                    self.outputs.append(s)  # 将客户端的socket添加到outputs列表

            for w in writeable:  # 客户端写列表
                data = self.msg[w]['queue'].get()
                # print(data)
                # 判断账号是否登录，未登录执行如下
                if self.msg[w]['status'] == False:
                    self.msg[w]['account'].append(data.decode())
                    if len(self.msg[w]['account']) == 2:
                        # 进行账号密码验证操作
                        out = self.login_auth(self.msg[w]['account'][0],self.msg[w]['account'][1],setting.account)
                        if out == '200':
                            self.msg[w]['status'] = True
                            w.send('login'.encode('utf-8'))
                        else:
                            w.send('err'.encode('utf-8'))
                            exceptionable.append(w)
                else:
                # 已登录执行如下
                # 根据客户端socket进行get和put操作
                    if w in self.client_sock_cmd:
                        if self.client_sock_cmd[w]['cmd'] == b'put':
                            # 进行put上传操作
                            self.client_sock_cmd[w]['file_io'].write(data)
                            self.client_sock_cmd[w]['file_io'].flush()
                            if os.path.getsize(self.client_sock_cmd[w]['filename']) == self.client_sock_cmd[w]['filesize']:
                                self.client_sock_cmd[w]['file_io'].close()
                                del self.client_sock_cmd[w]
                                print('文件保存完毕')
                        else:
                            # 进行get下载操作
                            if data.decode() == self.client_sock_cmd[w]['status']:
                                # 进行数据get操作在Linux必须要用send指定字节大小，否则报错，在windows下可以一次性发送sendall
                                # 由于这里使用了select多路复用模式，所以需要每次收发，都需要和客户端进行一次交互动作
                                if self.client_sock_cmd[w]['filesize'] - self.client_sock_cmd[w]['initsize'] >= 1024:
                                    size = 1024
                                else:
                                    size = self.client_sock_cmd[w]['filesize'] - self.client_sock_cmd[w]['initsize']
                                data = self.client_sock_cmd[w]['file_io'].read(size)
                                self.client_sock_cmd[w]['initsize'] += len(data)
                                w.send(data)
                                if self.client_sock_cmd[w]['filesize'] == self.client_sock_cmd[w]['initsize']:
                                    self.client_sock_cmd[w]['file_io'].close()
                                    del self.client_sock_cmd[w]
                                    print('文件发送完毕')
                    else:
                        # 根据客户端的socket进行字典初始化
                        self.client_sock_cmd[w] = {}
                        self.client_sock_cmd[w]['cmd'] = data.split()[0]
                        if self.client_sock_cmd[w]['cmd'] == b'put':
                            # 进行put的初始化操作
                            self.client_sock_cmd[w]['filename'], self.client_sock_cmd[w]['filesize'] = data.split()[1], int(data.split()[2])
                            self.client_sock_cmd[w]['filename'] = os.path.join(setting.save_dir,self.client_sock_cmd[w]['filename'])
                            f = open(self.client_sock_cmd[w]['filename'], 'wb')
                            self.client_sock_cmd[w]['file_io'] = f
                            w.send('200'.encode('utf-8'))
                        else:
                            # 进行get的初始化操作
                            self.client_sock_cmd[w]['filename'] = data.split()[1]
                            self.client_sock_cmd[w]['filename'] = os.path.join(setting.save_dir,self.client_sock_cmd[w]['filename'])
                            if os.path.isfile(self.client_sock_cmd[w]['filename']):
                                self.client_sock_cmd[w]['status'] = '200'
                                self.client_sock_cmd[w]['initsize'] = 0
                                self.client_sock_cmd[w]['filesize'] = os.path.getsize(self.client_sock_cmd[w]['filename'])
                                status_msg = '%s %s' % (self.client_sock_cmd[w]['status'], str(self.client_sock_cmd[w]['filesize']))
                                f = open(self.client_sock_cmd[w]['filename'], 'rb')
                                self.client_sock_cmd[w]['file_io'] = f
                            else:
                                status_msg = '%s 0' % ('404')
                                exceptionable.append(w)
                            w.send(status_msg.encode('utf-8'))
                self.outputs.remove(w)
            # 客户端异常或者退出清空对应的客户端socket信息并关闭连接
            for e in exceptionable:
                print('客户端退出:',e)
                if e in self.msg:
                    del self.msg[e]
                if e in self.outputs:
                    self.outputs.remove(e)
                self.inputs.remove(e)
                e.close()

客户端核心代码

#!/bin/env python
#Author: zhaoyong

import socket,os


class client_sock(object):
    '''
        ftp客户端主程序
    '''

    def __init__(self):
        ''' socket 实例化'''
        self.client = socket.socket()

    def conn(self,ip,port):
        ''' 连接服务器'''
        self.client.connect((ip,port),)

    def help(self):
        ''' 帮助信息'''
        print('''
            上传下载命令如下：
            put filename  上传文件
            get filename  下载文件
        ''')

    def interactive(self):
        ''' 交互模式'''
        while True:
            msg = self.client.recv(22).decode()
            if msg == 'login':
                print('登录成功')
                break
            if msg == 'err':
                exit('登录失败')
            info = input(msg).strip()
            if not info:
                exit('输入为空，退出')
            self.client.send(info.encode('utf-8'))
        while True:
            cmd = input('请输入上传下载命令<exit 退出系统>:').strip()
            if not cmd:
                print('输入为空，请重新输入')
                continue
            if cmd == 'exit':
                print('客户端退出完成')
                self.client.send(b'exit')
                break
            if len(cmd.split()) == 2:
                # 反射调用类方法
                if hasattr(self,cmd.split()[0]):
                    fun = getattr(self,cmd.split()[0])
                    fun(cmd)
                else:
                    self.help()
            else:
                self.help()

    def put(self,cmd):
        ''' 客户端执行的put功能'''
        # 提取文件名
        file_name = cmd.split()[1].strip()
        if os.path.isfile(file_name):
            # 提取上传的文件大小
            file_size = os.path.getsize(file_name)
            file_mess = '%s %s' % (cmd, str(file_size))
            # 将文件命令，文件名，文件大小发送到服务器
            self.client.send(file_mess.encode('utf-8'))
            # 打开文件读取并发送
            mess = self.client.recv(1024).decode()
            if mess == '200':
                f = open(file_name, 'rb')
                self.client.sendall(f.read())
                f.close()
                print('文件上传完毕')
        else:
            print('文件不存在')

    def get(self,cmd):
        ''' 客户端执行的get功能'''
        self.client.send(cmd.encode('utf-8'))
        # 以下为接收到服务器发来的文件名和文件大小
        file_name = cmd.split()[1]
        file_mess = self.client.recv(1024).decode()
        file_mess = file_mess.split()
        file_status, file_size = file_mess[0], int(file_mess[1])
        if file_status == '404':
            print('文件不存在')
            return 0
        # 接收到服务器端的状态码后，返回给服务器端确认
        self.client.send(file_status.encode('utf-8'))
        # 打开写文件
        init_size = 0
        f = open(file_name, 'wb')
        while init_size < file_size:
            if file_size - init_size >= 1024:
                size = 1024
            else:
                size = file_size - init_size
            data = self.client.recv(size)
            init_size += len(data)
            f.write(data)
            f.flush()
            # 如果文件没有下载完成，就重复通知服务器
            if init_size < file_size:
                self.client.send(file_status.encode('utf-8'))
        if os.path.getsize(file_name) == file_size:
            f.close()
            print('文件下载完毕')

关于selectors 模块

举例：

import socket
import selectors

sel = selectors.DefaultSelector()

def accept(sock,mask):
    conn,addr = sock.accept()
    print('fount client',conn)
    conn.setblocking(False)
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    data = conn.recv(1024)
    print(data)
    if data:
        conn.send(data)
    else:
        sel.unregister(conn)
        conn.close()

server = socket.socket()
server.bind(('0.0.0.0',9999))
server.listen(100)
server.setblocking(False)
sel.register(server,selectors.EVENT_READ,accept)
while True:
    events = sel.select()
    for key,mask in events:
        callback = key.data
        callback(key.fileobj, mask)