python全栈开发基础【第二十七篇】IO模型

#所谓同步，就是在发出一个功能调用时，在没有得到结果之前，该调用就不会返回。按照这个定义，
其实绝大多数函数都是同步调用。但是一般而言，我们在说同步、异步的时候，
特指那些需要其他部件协作或者需要一定时间完成的任务。
#举例：
#1. multiprocessing.Pool下的apply #发起同步调用后，就在原地等着任务结束，
根本不考虑任务是在计算还是在io阻塞，总之就是一股脑地等任务结束
#2. concurrent.futures.ProcessPoolExecutor().submit(func,).result()
#3. concurrent.futures.ThreadPoolExecutor().submit(func,).result()

#阻塞调用是指调用结果返回之前，当前线程会被挂起（如遇到io操作）。函数只有在得到结果之后才会
将阻塞的线程激活。有人也许会把阻塞调用和同步调用等同起来，实际上他是不同的。对于同步调用来说，
很多时候当前线程还是激活的，只是从逻辑上当前函数没有返回而已。
#举例：
#1. 同步调用：apply一个累计1亿次的任务，该调用会一直等待，直到任务返回结果为止，
但并未阻塞住（即便是被抢走cpu的执行权限，那也是处于就绪态）;
#2. 阻塞调用：当socket工作在阻塞模式的时候，如果没有数据的情况下调用recv函数，
则当前线程就会被挂起，直到有数据为止。

#服务端
from socket import *
server = socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind((‘127.0.0.1‘,8080))
server.listen(5)
print(‘start runnig...‘)
while True:
    conn,addr = server.accept()  #IO操作 在这accept的时候不能干recv的活
    print(addr)
    while True:
        try:
            data = conn.recv(1024)  #IO操作
            conn.send(data.upper())
        except Exception:
            break
    conn.close()
server.close()

# 我们以前写的这个就是阻塞的IO模型：一旦阻塞了就在那卡着
# 直到等到数据已经到了操作系统，操作系统再从内核拷贝给应用程序
# 阻塞IO在那两个阶段全都阻塞住了

#客户端
from socket import *
client  = socket(AF_INET,SOCK_STREAM)
client.connect((‘127.0.0.1‘,8080))
while True:
    cmd = input(‘>>:‘).strip()
    if not cmd:continue
    client.send(cmd.encode(‘utf-8‘))
    data = client.recv(1024)
    print(‘接受的是：%s‘%data.decode(‘utf-8‘))
client.close()

#这种程序虽说解决了单线程并发，但是大大的占用了cpu
from socket import *
import time
severt = socket(AF_INET,SOCK_STREAM)
severt.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
severt.bind((‘127.0.0.1‘,8080))
severt.listen(5)
severt.setblocking(False) #默认是True  (如果是False,套接字里的一些阻塞操作都变成非阻塞的)
print(‘startting....‘)
conn_l = []
del_l =[]
while True:
    try:
        print(conn_l)
        conn,addr = severt.accept() #收不到数据的时候才出异常
        print(conn)
        conn_l.append(conn)
    except BlockingIOError:  #吧收不到数据的那段时间利用起来（利用他收不到
        #数据的时候，才干下面的for循环）
       for conn in conn_l:
            try:
                data = conn.recv(1024)
                conn.send(data.upper())
            except BlockingIOError:
                pass
            except ConnectionResetError:  #端开链接的错误（如果突然断开链接，会报错
                                           #就先添加到列表里面去，完了吧链接给清除了）
                 del_l.append(conn)
       for obj in del_l:
             obj.close()
             conn_l.remove(obj)
       del_l.clear()

from socket import *
client = socket(AF_INET,SOCK_STREAM)
client.connect((‘127.0.0.1‘,8080))
while True:
    cmd = input(‘>>:‘).strip()
    if not cmd:continue
    client.send(cmd.encode(‘utf-8‘))
    data = client.recv(1024)
    print(data.decode(‘utf-8‘))

#select IO模型
#服务端 
from socket import *
import select

s=socket(AF_INET,SOCK_STREAM)
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind((‘127.0.0.1‘,8081))
s.listen(5)
s.setblocking(False) #设置socket的接口为非阻塞
read_l=[s,]
while True:
    r_l,w_l,x_l=select.select(read_l,[],[])
    print(r_l)
    for ready_obj in r_l:
        if ready_obj == s:
            conn,addr=ready_obj.accept() #此时的ready_obj等于s
            read_l.append(conn)
        else:
            try:
                data=ready_obj.recv(1024) #此时的ready_obj等于conn
                if not data:
                    read_l.remove(ready_obj)
                    continue
                ready_obj.send(data.upper())
            except ConnectionResetError:
                read_l.remove(ready_obj)

#客户端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect((‘127.0.0.1‘,8081))

while True:
    msg=input(‘>>: ‘)
    if not msg:continue
    c.send(msg.encode(‘utf-8‘))
    data=c.recv(1024)
    print(data.decode(‘utf-8‘))

#服务端（多路复用IO）
# select模块用select方法检测那个套接字准备好了，也就是收没收到数据（而我们的
# 非阻塞IO你不知道那个套接字准备好了，那么用select模块就能解决
# 这个问题）
# select还可以检测多个套接字
# 所以select比非阻塞IO的效率高
from socket import *
import select
server = socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind((‘127.0.0.1‘,8081))
server.setblocking(False) #设置socket的套接字为非阻塞的
server.listen(5)
print(‘start running....‘)
read_l = [server,]  #因为不只就那么一个列表要检测。所以不要在参数里面定死了
while True:
    r_l,w_l,x_l = select.select(read_l,[],[])  #select()方法有四个参数
    print(r_l)  #一开始服务端运行的时候，就等着，当你客户端一链接的时候，他就
                # 检测到有数据了(检测那个数据准备好了)
    for obj in r_l:
        if obj == server:
            conn,addr = obj.accept()  #accept要经历两个阶段，但是程序如果走到这一步，那肯定是数据准备好了
                         #当数据已经准备好的时候，accept就只经历一个copy数据的阶段了
            # print(addr)
            read_l.append(conn)  #在监听一下conn套接字（这时候已经监听了两个了：分别是accept,conn）
        else:
            data = obj.recv(1024)  # 此时的obj=conn
            obj.send(data.upper())
#         obj.close()
# server.close()

#客户端（多路复用IO）
from socket import *
import select
client = socket(AF_INET,SOCK_STREAM)
client.connect((‘127.0.0.1‘,8081))
while True:
    cmd = input(‘>>:‘)
    client.send(cmd.encode(‘utf-8‘))
    data = client.recv(1024)
    print(‘接收的是：%s‘%data.decode(‘utf-8‘))
client.close()

# selectors 
# 服务端
from socket import *
import selectors

sel=selectors.DefaultSelector()
def accept(server_fileobj,mask):
    conn,addr=server_fileobj.accept()
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print(‘closing‘,conn)
            sel.unregister(conn)
            conn.close()
            return
        conn.send(data.upper()+b‘_SB‘)
    except Exception:
        print(‘closing‘, conn)
        sel.unregister(conn)
        conn.close()



server_fileobj=socket(AF_INET,SOCK_STREAM)
server_fileobj.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server_fileobj.bind((‘127.0.0.1‘,8088))
server_fileobj.listen(5)
server_fileobj.setblocking(False) #设置socket的接口为非阻塞
sel.register(server_fileobj,selectors.EVENT_READ,accept) #相当于网select的读列表里append了一个文件句柄server_fileobj,并且绑定了一个回调函数accept

while True:
    events=sel.select() #检测所有的fileobj，是否有完成wait data的
    for sel_obj,mask in events:
        callback=sel_obj.data #callback=accpet
        callback(sel_obj.fileobj,mask) #accpet(server_fileobj,1)

#客户端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect((‘127.0.0.1‘,8088))

while True:
    msg=input(‘>>: ‘)
    if not msg:continue
    c.send(msg.encode(‘utf-8‘))
    data=c.recv(1024)
    print(data.decode(‘utf-8‘))