1.IO概述
input 和 output: 是在内存中存在的数据交换操作
内存和磁盘交换: 文件读写, 打印
内存和网络交换: recv send recvfrom, sendto
IO密集型程序: 程序中执行大量的IO操作,而较少的需要CPU运行,消耗CPU资源少,运行周期往往比较长
CPU密集型程序: 程序中执行大量的CPU运算,而较少的需要IO操作,消耗CPU资源多
2.IO分类(阻塞/非阻塞)
阻塞IO: 默认形态,是效率最低的一种IO
1.因为等待某种条件达成再继续运行,例如: accept recv input
2.处理IO事件时耗时较长也会产生阻塞,例如: 文件的读写过程,网络数据的传输过程
非阻塞IO: 通过修改IO对象使其变为非阻塞状态,通常用循环来不断判断阻塞条件,需要消耗更多的CPU但是在一定程度上提高了IO效率
IO多路复用: 通过同时监控多个IO事件,当那个IO事件就绪就执行那个IO事件,形成并发效果,也被称作事件驱动IO
信号驱动IO: 准备数据阶段(用户进程非阻塞),复制数据阶段(用户进程阻塞)
1.当调用read函数的时候,准备数据的过程中用户线程不阻塞,用户线程可以去做其他事情
2.等到数据准备完了,用户进程会收到一个SIGIO信号,然后可以在信号处理函数中处理数据
异步IO: Python实现不了,但是tornado框架自带异步IO
1.阻塞IO,非阻塞IO,多路复用IO,信号驱动IO都是同步IO,都需等待将数据从内核缓冲区拷贝到用户内存,会阻塞进程
2.异步IO是异步IO,当用户发起一个操作后就立即去做其他事情了,内核会等待,检测数据直到数据准备完成
3.然后将数据拷贝到用户内存(无需先拷贝到内核缓冲区),这一切完成后会给用户进程发送一个信号告知操作完成了
3.非阻塞IO-效率比阻塞IO高
1.非阻塞IO设置方法
import socket s = socket.socket() # 默认会创建流式套接字 # 修改套接字设置的阻塞状态 # 参数(bool类型): 默认为True,设置为False则为非阻塞 s.setblocking()
2.TCP服务端-非阻塞IO
import socket import time def main(): # 创建数据流套接字 tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置端口重用 tcp_server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 绑定端口,运行局域网内的客户端连接 tcp_server_socket.bind(("0.0.0.0", 7890)) # 让默认的套接字由主动变为被动监听 listen tcp_server_socket.listen(128) # 设置套接字为非阻塞状态 tcp_server_socket.setblocking(False) # 循环目的: 多次调用accept等待客户端连接,为多个客服端服务 while True: print("等待一个新的客户端的到来...") try: new_client_socket, client_addr = tcp_server_socket.accept() except BlockingIOError: time.sleep(1) print(time.ctime()) continue print("客户端' %s '已经到来" % str(client_addr)) # 设置客户端字为非阻塞状态 new_client_socket.setblocking(False) # 循环目的: 为同一个客服端服务多次 while True: try: # 接收客户端发送过来的请求 recv_data = new_client_socket.recv(1024) # recv解堵塞的两种方式: 1.客户端发送过来数据,2.客户端调用了close if recv_data: print("客户端' %s '发送过来的请求是: %s" % (str(client_addr), recv_data.decode("utf-8"))) # 回送数据给客户端表示响应客户端的请求 send_data = "----ok----Request accepted" new_client_socket.send(send_data.encode("utf-8")) else: break except BlockingIOError: time.sleep(1) print(time.time()) continue # 关闭accept返回的套接字 new_client_socket.close() print("已经为 %s 客户端已经服务完毕" % str(client_addr)) # 关闭监听套接字 tcp_server_socket.close() if __name__ == "__main__": main()
3.TCP客户端-非阻塞IO
import socket def main(): # 创建数据流套接字 tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 连接服务器 server_ip = input("请输入要连接的服务器的ip:") serve_port = int(input("请输入要连接的服务器的port:")) server_addr = (server_ip, serve_port) tcp_client_socket.connect(server_addr) # 发送数据 send_data = input("请输入要发生的数据:") tcp_client_socket.send(send_data.encode("utf-8")) # 接收服务器发送过来的数据 recv_data = tcp_client_socket.recv(1024) print("接收到的数据为:%s" % recv_data.decode("utf-8")) # 关闭套接字 tcp_client_socket.close() if __name__ == "__main__": main()
4.超时检测(超时等待)-效率比非阻塞IO高
1.超时检测设置
import socket s = socket.socket() # 默认会创建流式套接字 # 设置套接字的超时检测,所谓的超时检测即对原本阻塞的函数进行设置,使其不再始终阻塞,而是阻塞等待一定时间后自动返回 # 参数: 超时时间,在规定时间中如果正常接收阻塞则继续执行否则产生timeout异常 s.settimeout(5) # 设置超时时间为5秒
2.TCP服务端-超时检测
import socket import traceback def main(): # 创建数据流套接字 tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置端口重用 tcp_server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 绑定端口,运行局域网内的客户端连接 tcp_server_socket.bind(("0.0.0.0", 7890)) # 让默认的套接字由主动变为被动监听 listen tcp_server_socket.listen(128) # 设置套接字超时等待时间为5秒 tcp_server_socket.settimeout(5) # 循环目的: 多次调用accept等待客户端连接,为多个客服端服务 while True: print("等待一个新的客户端的到来...") try: new_client_socket, client_addr = tcp_server_socket.accept() except Exception: traceback.print_exc() continue print("客户端' %s '已经到来" % str(client_addr)) # 设置客户端字的超时等待时间为5秒 new_client_socket.settimeout(5) # 循环目的: 为同一个客服端服务多次 while True: try: # 接收客户端发送过来的请求 recv_data = new_client_socket.recv(1024) # recv解堵塞的两种方式: 1.客户端发送过来数据,2.客户端调用了close if recv_data: print("客户端' %s '发送过来的请求是: %s" % (str(client_addr), recv_data.decode("utf-8"))) # 回送数据给客户端表示响应客户端的请求 send_data = "----ok----Request accepted" new_client_socket.send(send_data.encode("utf-8")) else: break except Exception: traceback.print_exc() continue # 关闭accept返回的套接字 new_client_socket.close() print("已经为 %s 客户端已经服务完毕" % str(client_addr)) # 关闭监听套接字 tcp_server_socket.close() if __name__ == "__main__": main()
3.TCP客户端-超时检测
import socket def main(): # 创建数据流套接字 tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 连接服务器 server_ip = input("请输入要连接的服务器的ip:") serve_port = int(input("请输入要连接的服务器的port:")) server_addr = (server_ip, serve_port) tcp_client_socket.connect(server_addr) # 发送数据 send_data = input("请输入要发生的数据:") tcp_client_socket.send(send_data.encode("utf-8")) # 接收服务器发送过来的数据 recv_data = tcp_client_socket.recv(1024) print("接收到的数据为:%s" % recv_data.decode("utf-8")) # 关闭套接字 tcp_client_socket.close() if __name__ == "__main__": main()
5.IO多路复用-效率比超时等待高
1.IO多路复用概述
1.IO多路复用的定义: 同时监控多个IO事件,当那个IO事件就绪就执行那个IO事件,形成并发效果
2.使用IO多路复用的注意点
1.在处理IO过程中不应该发生死循环(即某个IO单独占用服务器)
2.IO多路复用是单进程程序,是一个并发程序
3.IO多路复用有较高的IO执行效率
2.IO多路复用-select模块中的三个方法(select,pool,epool)
1.三个方法的适用平台
select: 适用于Windows Linux Unix MacOS
poll: 适用于Linux Unix MacOS
epoll: 适用于Linux Unix
2.select方法-采用轮询机制,32位机文件描述符只能开1024,64位机文件描述符只能开2048
r,w,x = select(rlist, wlist, xlist, [timeout])
功能: 监控IO事件,阻塞等待IO事件发生
参数:
rlist: 列表,存放要监控等待处理的IO,例如客户端主动连接,或客户端发送消息
wlist: 列表,存放希望主动处理的IO,例如回送消息给客户端
xlist: 列表,存放如果发生异常需要处理的IO
timeout: 数字,超时检测,默认一直阻塞
返回值:
r: 列表,rlist中准备就绪的IO
w: 列表,wlist中准备就绪的IO
x: 列表,xlist中准备就绪的IO
3.poll方法-采用轮询机制突破了文件描述符只能开1024的限制(1G内存大概能开10万个事件去监听),效率上和select差不多
1.创建poll对象
p = select.poll()
2.加入关注的IO
p.register(s) # 添加IO关注
p.unregister(s) # 从关注IO中删除
3.使用poll函数监控
events = p.poll()
功能: 阻塞等待register的事件只要有任意准备就绪即返回
返回值: events # [(fileno, event), (), ()]; fileno: 文件描述符; event: 准备就绪的事件
4.处理发生的IO事件
# poll的IO事件
POLLIN # 可读rlist
POLLOUT # 可写wlist
POLLUP # 断开连接
POLLERR # 异常xlist
POLLPRI # 紧急处理
POLLVAL # 无效数据
4.epoll方法: 采用了回调机制同样也突破了文件描述符只能开1024的限制(1G内存大概能开10万个事件去监听)
1.效率上比poll和select稍高,但只能用于Linux和Unix平台
2.epoll既可以采用水平触发,也可以采用边缘触发,而select和pool只支持水平触发
3.epoll实现原理提流程图: https://www.processon.com/view/link/5efd699c7d9c08442043e5b8
4.水平触发: 如果文件描述符已经就绪可以非阻塞的执行IO操作了,此时会触发通知,允许在任意时刻重复检测IO的状态,
没有必要每次描述符就绪后尽可能多的执行IO,select和pool就属于水平触发
5.边缘触发: 如果文件描述符自上次状态改变后有新的IO活动到来,此时会触发通知,在收到一个IO事件通知后要尽可能多的执行IO操作,
因为如果在一次通知中没有执行完IO那么就需要等到下一次新的IO活动到来才能获取到就绪的描述符,信号驱动式IO就属于边缘触发
3.TCP服务端-select方法实现IO多路复用
import socket import sys import traceback import select def main(): # 创建数据流套接字 tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置端口重用 tcp_server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 绑定端口,运行局域网内的客户端连接 tcp_server_socket.bind(("0.0.0.0", 7890)) # 让默认的套接字由主动变为被动监听 listen tcp_server_socket.listen(128) # 设置套接字超时等待时间为5秒 tcp_server_socket.settimeout(5) # 创建一个列表存放要监控等待处理的IO rlist = [tcp_server_socket] # 创建一个列表存放希望主动处理的IO wlist = list() # 创建一个列表存放如果发生异常需要处理的IO xlist = [tcp_server_socket] # 循环目的: 多次调用select监控客户端连接事件,阻塞等待客户端连接事件发生 while True: print("监控是否有新的客户端的到来...") # 参数说明: rlist: 列表,存放要监控等待处理的IO; wlist: 列表,存放希望主动处理的IO # 参数说明: xlist: 列表,存放如果发生异常需要处理的IO; timeout: 数字,超时检测,默认一直阻塞 # 返回值说明: rs: 列表,rlist中准备就绪的IO; ws: 列表,wlist中准备就绪的IO; xs: 列表,xlist中准备就绪的IO # wlist有内容,select会解除阻塞立即返回 rs, ws, xs = select.select(rlist, wlist, xlist, 5) # 设置超时等待时间为5秒 # 遍历准备就绪的IO列表 for r in rs: # 有客户端连接时 if r is tcp_server_socket: new_client_socket, client_addr = r.accept() print("客户端' %s '已经到来" % str(client_addr)) # 将为客户端服务的套接字加入监控列表rlist rlist.append(new_client_socket) else: try: # 已经连接的客户端发送消息时 recv_data = r.recv(1024) if recv_data: # r.getsockname()可以获取套接字绑定的地址 print("客户端' %s '发送过来的请求是: %s" % (str(r.getsockname()), recv_data.decode("utf-8"))) # 服务器回送消息放到主动处理列表wlist wlist.append(r) else: # 如果客户端是调用了close断开了连接,那么需要从监控列表中删除为客户端服务的套接字 rlist.remove(r) print("已经为 %s 客户端已经服务完毕" % str(r.getsockname())) # 关闭为客户端服务的套接字 r.close() except Exception: traceback.print_exc() # 遍历需要服务器主动处理的IO列表 for w in ws: # 回送数据给客户端表示响应客户端的请求 send_data = "----ok----Request accepted" w.send(send_data.encode("utf-8")) # 把为客户端服务的套接字从主动处理的列表中删除 wlist.remove(w) # 遍历异常请求的IO列表 for x in xs: # 如果监听套接字发生了异常 if x is tcp_server_socket: # 关闭异常的监听套接字 x.close() # 退出程序 sys.exit(1) if __name__ == "__main__": main()
4.TCP客户端-select方法实现IO多路复用
import socket def main(): # 创建数据流套接字 tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 连接服务器 server_ip = input("请输入要连接的服务器的ip:") serve_port = int(input("请输入要连接的服务器的port:")) server_addr = (server_ip, serve_port) tcp_client_socket.connect(server_addr) # 发送数据 send_data = input("请输入要发生的数据:") tcp_client_socket.send(send_data.encode("utf-8")) # 接收服务器发送过来的数据 recv_data = tcp_client_socket.recv(1024) print("接收到的数据为:%s" % recv_data.decode("utf-8")) # 关闭套接字 tcp_client_socket.close() if __name__ == "__main__": main()
5.TCP服务器-pool方法实现IO多路复用
import socket import select import traceback import sys def main(): # 创建数据流套接字 tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置端口重用 tcp_server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 绑定端口,运行局域网内的客户端连接 tcp_server_socket.bind(("0.0.0.0", 7890)) # 让默认的套接字由主动变为被动监听 listen tcp_server_socket.listen(128) # 设置套接字超时等待时间为5秒 tcp_server_socket.settimeout(5) # 创建一个IO事件地图 fdmap = {tcp_server_socket.fileno(): tcp_server_socket} # 创建poll对象 epool_list = select.poll() # 将监听套接对应的fd注册到poll中进行监听,如果fd已经注册过,则会发生异常 epool_list.register(tcp_server_socket, (select.POLLIN | select.POLLERR)) # 循环目的: 等待新的客户端连接或者已连接的客户端发送过来数据 while True: print("开始事件监控") events = epool_list.poll() for fd, event in events: if fd == tcp_server_socket.fileno(): new_client_socket, client_addr = tcp_server_socket.accept() print("客户端' %s '已经到来" % str(client_addr)) epool_list.register(new_client_socket, select.POLLIN) # 记录这个信息 fdmap[new_client_socket.fileno()] = new_client_socket elif event & select.POLLIN: try: # 已经连接的客户端发送消息时 recv_data = fdmap[fd].recv(1024) if not recv_data: # 在poll中注销客户端的信息 epool_list.unregister(fd) print("已经为客户端已经服务完毕") # 关闭客户端的文件句柄 fdmap[fd].close() # 在字典中删除与已关闭客户端相关的信息 del fdmap[fd] else: print("客户端发送过来的请求是: %s" % (recv_data.decode("utf-8"))) # 回送数据给客户端表示响应客户端的请求 send_data = "----ok----Request accepted" fdmap[fd].send(send_data.encode("utf-8")) except Exception: traceback.print_exc() elif event & select.POLLERR: # 如果监听套接字发生了异常 if fd is tcp_server_socket.fileno(): # 关闭异常的监听套接字 fdmap[fd].close() # 退出程序 sys.exit(1) if __name__ == "__main__": main()
6.TCP客户端-pool方法实现IO多路复用
import socket def main(): # 创建数据流套接字 tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 连接服务器 server_ip = input("请输入要连接的服务器的ip:") serve_port = int(input("请输入要连接的服务器的port:")) server_addr = (server_ip, serve_port) tcp_client_socket.connect(server_addr) # 发送数据 send_data = input("请输入要发生的数据:") tcp_client_socket.send(send_data.encode("utf-8")) # 接收服务器发送过来的数据 recv_data = tcp_client_socket.recv(1024) print("接收到的数据为:%s" % recv_data.decode("utf-8")) # 关闭套接字 tcp_client_socket.close() if __name__ == "__main__": main()
7.TCP服务器-epool方法实现IO多路复用
import socket import select import traceback import sys def main(): # 创建数据流套接字 tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置端口重用 tcp_server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 绑定端口,运行局域网内的客户端连接 tcp_server_socket.bind(("0.0.0.0", 7890)) # 让默认的套接字由主动变为被动监听 listen tcp_server_socket.listen(128) # 设置套接字超时等待时间为5秒 tcp_server_socket.settimeout(5) # 创建一个IO事件地图 fdmap = {tcp_server_socket.fileno(): tcp_server_socket} # 创建poll对象 epool_list = select.poll() # 将监听套接对应的fd注册到poll中进行监听,如果fd已经注册过,则会发生异常 epool_list.register(tcp_server_socket, (select.EPOLLIN | select.EPOLLERR)) # 循环目的: 等待新的客户端连接或者已连接的客户端发送过来数据 while True: print("开始事件监控") events = epool_list.poll() for fd, event in events: if fd == tcp_server_socket.fileno(): new_client_socket, client_addr = tcp_server_socket.accept() print("客户端' %s '已经到来" % str(client_addr)) epool_list.register(new_client_socket, select.EPOLLIN) # 记录这个信息 fdmap[new_client_socket.fileno()] = new_client_socket elif event & select.EPOLLIN: try: # 已经连接的客户端发送消息时 recv_data = fdmap[fd].recv(1024) if not recv_data: # 在poll中注销客户端的信息 epool_list.unregister(fd) print("已经为客户端已经服务完毕") # 关闭客户端的文件句柄 fdmap[fd].close() # 在字典中删除与已关闭客户端相关的信息 del fdmap[fd] else: print("客户端发送过来的请求是: %s" % (recv_data.decode("utf-8"))) # 回送数据给客户端表示响应客户端的请求 send_data = "----ok----Request accepted" fdmap[fd].send(send_data.encode("utf-8")) except Exception: traceback.print_exc() elif event & select.EPOLLERR: if fd is tcp_server_socket.fileno(): # 关闭异常的监听套接字 fdmap[fd].close() # 退出程序 sys.exit(1) if __name__ == "__main__": main()
8.TCP客户端-epool方法实现IO多路复用
import socket def main(): # 创建数据流套接字 tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 连接服务器 server_ip = input("请输入要连接的服务器的ip:") serve_port = int(input("请输入要连接的服务器的port:")) server_addr = (server_ip, serve_port) tcp_client_socket.connect(server_addr) # 发送数据 send_data = input("请输入要发生的数据:") tcp_client_socket.send(send_data.encode("utf-8")) # 接收服务器发送过来的数据 recv_data = tcp_client_socket.recv(1024) print("接收到的数据为:%s" % recv_data.decode("utf-8")) # 关闭套接字 tcp_client_socket.close() if __name__ == "__main__": main()