一, 阻塞 IO(blocking IO)
- from concurrent.futures import ThreadPoolExecutor
- import socket
- server = socket.socket()
- # 重用端口
- server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
- server.bind(("127.0.0.1",9999))
- server.listen(5)
- # 线程池
- pool = ThreadPoolExecutor(3)
- def data_handler(conn):
- print("一个新连接..")
- while True:
- data = conn.recv(1024)
- conn.send(data.upper())
- while True:
- conn,addr = server.accept()
- # 切到处理数据的任务去执行
- pool.submit(data_handler,conn)
线程池阻塞 IO 服务端
- import socket
- c = socket.socket()
- c.connect(("127.0.0.1",9999))
- while True:
- msg = input(">>>:")
- if not msg:continue
- c.send(msg.encode("utf-8"))
- data = c.recv(1024)
- print(data.decode("utf-8"))
线程池阻塞 IO 客服端
二, 非阻塞 IO(non-blocking IO)
- from concurrent.futures import ThreadPoolExecutor
- import socket
- server = socket.socket()
- # 重用端口
- server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
- server.bind(("192.168.11.210",9999))
- server.listen(5)
- # 设置是否为阻塞 默认阻塞
- server.setblocking(False)
- def data_handler(conn):
- print("一个新连接..")
- while True:
- data = conn.recv(1024)
- conn.send(data.upper())
- # 已连接的客户端
- clients = []
- # 需要发送的数据
- send_datas = []
- # 已经发送完的 需要删除的数据
- del_datas = []
- # 待关闭的客户端
- closed_cs = []
- import time
- while True:
- try:
- conn,addr = server.accept()
- # 切到处理数据的任务去执行
- # 代码走到这里才算是连接成功
- # 把连接成功的客户端存起来
- clients.append(conn)
- except BlockingIOError:
- # print("没有可以处理的连接 就干别的活儿")
- #要处理的是已经连接成功的客户端
- # 接收数据
- for c in clients:
- try:
- data = c.recv(1024)
- if not data:
- # 对方关闭了连接
- c.close()
- # 从客户端列表中删除它
- closed_cs.append(c)
- continue
- print("收到 %s" % data.decode("utf-8"))
- # 现在非阻塞 send 直接往缓存赛 如果缓存满了 肯定有错误 需要单独处理发送
- # c.send(data.upper())
- send_datas.append((c,data))
- except BlockingIOError:
- pass
- except ConnectionResetError:
- # 对方关闭了连接
- c.close()
- # 从客户端列表中删除它
- closed_cs.append(c)
- # 处理发送数据
- for data in send_datas:
- try:
- data[0].send(data[1].upper())
- # 发送成功需要删除 不能直接删除
- # send_datas.remove(data)
- del_datas.append(data)
- except BlockingIOError:
- continue
- except ConnectionResetError:
- # 客户端连接需要删除
- data[0].close()
- closed_cs.append(data[0])
- # 等待发送的数据需要删除
- del_datas.append(data)
- # 删除无用的数据
- for d in del_datas:
- #从待发送的列表中删除
- send_datas.remove(d)
- del_datas.clear()
- for c in closed_cs:
- clients.remove(c)
- closed_cs.clear()
服务器
- import socket
- c = socket.socket()
- c.connect(("127.0.0.1",9999))
- while True:
- msg = input(">>>:")
- if not msg:continue
- c.send(msg.encode("utf-8"))
- data = c.recv(1024)
- print(data.decode("utf-8"))
客服端
- li = [1,2,3,4,5,6]
- def mytlist_iter():
- for i in range(len(li)):
- yield li[i]
- for j in mytlist_iter():
- if j == 5:
- li.append(1000)
- d = {"a":1,"b":2}
- for k in d:
- if k == "a":
- d.pop(k)
迭代期间不能修改被迭代的对象
三, 多路复用 IO(IO multiplexing)
- from concurrent.futures import ThreadPoolExecutor
- import socket
- import select
- # select 帮你从一堆连接中找出来需要被处理的连接
- server = socket.socket()
- # 重用端口
- server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
- server.bind(("192.168.11.210",9999))
- server.listen(5)
- # 设置是否为阻塞 默认阻塞
- # server.setblocking(False)
- def data_handler(conn):
- print("一个新连接..")
- while True:
- data = conn.recv(1024)
- conn.send(data.upper())
- # 需要检测的 是否可读取的列表 (recv 就是一个读取操作)
- rlist = [server,]
- # 需要检测的 是否写入的列表 (send 就是写入操作)
- wlist = []
- # 需要发送的数据 目前是因为 我们要把接收的数据在发回去 所以搞了这个东西 正常没有这种需求
- # 目前客户端与服务器端 交互 是必须客户端发送数据 服务器端才能返回数据 正常没有这种需求
- dic = {}
- while True: # 用于检测需要处理的连接 需要不断检测 所以循环
- # rl 目前可读的客户端列表 wl 目前可写的客户端列表
- rl,wl,xl = select.select(rlist,wlist,[]) # select 默认阻塞 阻塞到任意一个连接可以被处理
- print(len(rl))
- # 处理可读的 socket
- for c in rl:
- # 无论是客户端还是服务器只要可读就会执行到这里
- if c == server:
- # 接收客户端的连接请求 (一个读操作)
- conn,addr = c.accept()
- # 将新连接也交给 select 来检测
- rlist.append(conn)
- else:# 不是服务器 就是客户端 客户端可读 可以执行 recv
- try:
- data = c.recv(1024)
- if not data:
- c.close()
- rlist.remove(c)
- print("%s 发送 %s" % (c,data.decode("utf-8")))
- # 给客户端发送数据 前要保证目前可以发送 将客户端加入检测列表
- wlist.append(c) # 正常开发中 不可能必须客户端发送数据过来后 才能 给客户端发送
- # 所以这个添加到检测列表的操作 应该建立连接后立即执行
- # 要发送的数据
- dic[c] = data
- except ConnectionResetError:
- # 客户端关闭连接
- c.close()
- rlist.remove(c)
- # 处理可写的 socket
- for c in wl:
- print(c)
- try:
- c.send(dic[c].upper())
- # 删除数据
- dic.pop(c)
- # 从检测列表中删除已发送完成的客户端
- wlist.remove(c)
- except ConnectionResetError:
- c.close() # 关闭连接
- dic.pop(c) # 删除要发送的数据
- wlist.remove(c) # 从待检测的列表中删除
- except BlockingIOError:# 可能缓存满了 发不了
- pass
服务器
- import socket
- c = socket.socket()
- c.connect(("192.168.11.210",9999))
- while True:
- msg = input(">>>:")
- if not msg:continue
- c.send(msg.encode("utf-8"))
- data = c.recv(1024)
- print(data.decode("utf-8"))
客户端
四, 异步 IO(Asynchronous I/O)
- import asyncio
- asyncio.coroutine()
- from concurrent.futures import ThreadPoolExecutor
- def task():
- print("read start")
- with open(r"D:\python 视频存放目录 \ 上海 python 全栈 4 期 \ day40 \ 多路复用, 降低 CPU 占用 \ 服务器. py",encoding="utf-8") as f:
- text = f.read()
- # f.write()
- print("read end")
- return text
- def fin(f):
- print("fin")
- print(f.result())
- pool = ThreadPoolExecutor(1)
- future = pool.submit(task)
- future.add_done_callback(fin)
- print("主 over")
- # 这种方式看起来像是异步 IO 但是对于子线程而言不是
- # 在子线程中 执行 read 是阻塞的 以为 CPU 必须切走 但是不能保证切到当前程序的其他线程
- # 想要的效果就是 在执行 read 是不阻塞 还能干其他活 谁能实现 只有协程
- # asyncio 内部是使用的是协程
模拟异步 IO
conclusion:
网络 IO 模型
一, 阻塞 IO 模型
多线程 多进程 线程池 进程池 全是阻塞 IO
二, 非阻塞 IO
协程是一种非阻塞 IO
1.setblocking(False) 将阻塞修改为非阻塞
2. 一旦是非阻塞 在执行 accept recv send 就会立马尝试读写数据 一旦数据没准备好就抛异常
3. 捕获异常
4. 如果没有异常说明数据准备好了 直接处理
5. 捕获到异常 那就做别的事情
可以实现单线程并发的效果 会大量占用 CPU 资源
三, 多路复用
将所有连接交给 select 来管理 管什么? 管哪个连接可以被处理
作为处理任务的一方事情变少了 不需要重复不断的问操作系统拿数据 而是等待 select 返回需要处理的连接, 等待则意味着 select 是阻塞的
异步 IO 不仅仅指网络 IO 也包括本地 IO
非阻塞 IO 和 多路复用 解决都是网络 IO 的阻塞问题
本地 IO 可以通过子线程 或子进程 来避免阻塞 但是对子线程或子进程而言 依然会阻塞
最终的解决方案就是协程 asyncio 该模快实现异步 IO 内部使用协程实现
来源: https://www.cnblogs.com/wanlei/p/9965416.html