I/O模型的使用

阻塞I/O

1.定义：在执行IO操作时如果执行条件不满足则阻塞。阻塞IO是IO的默认形态。

 

　　* 有那些I/O操作: 文件读写, 网络传输
　　* 阻塞: 进程由执行态进入阻塞态

2.效率：阻塞IO是效率很低的一种IO。但是由于逻辑简单所以是默认IO行为。

3.阻塞情况：

 

　　* 因为某种执行条件没有满足造成的函数阻塞

　　　　e.g.  input -- accept -- recv

　　* 处理IO的时间较长产生的阻塞状态

　　　　e.g. 网络传输，大文件读写

4. 代码

 

a = input()
print(a)

from socket import *
s = socket()
s.bind(('0.0.0.0', 8888))
s.listen(5)
print('Listen the port', 8888)
c, addr = s.accept()
print("Connect from", addr)

from socket import *
s = socket()
s.bind(('0.0.0.0', 8888))
s.listen(5)
print('Listen the port', 8888)
c, addr = s.accept()
print("Connect from", addr)

data = c.recv(1024)
print(data)

 

非阻塞I/O

1. 定义 ：通过修改IO属性行为，使原本阻塞的IO变为非阻塞的状态。

　　* 设置套接字为非阻塞IO

　　　　>sockfd.setblocking(False)　　-->  设置为非阻塞I/O后, 若产生阻塞, 则会报错: BlockingIOError

　　　　功能：设置套接字为非阻塞IO

　　　　参数：默认为True，表示套接字IO阻塞；设置为False则套接字IO变为非阻塞

 

　　* 超时检测 ：设置一个最长阻塞时间，超过该时间后则不再阻塞等待。

  　　　　>sockfd.settimeout(sec)　　--> 超时会引发_socket模块中的timeout异常

  　　　　功能：设置套接字的超时时间

  　　　　参数：设置的时间

 

"""
block_io.py
socket 非阻塞ＩＯ示例
"""

from socket import *
from time import *

# 日志文件
f = open('log.txt','a+')

# tcp　服务端
sockfd = socket()
sockfd.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
sockfd.bind(('0.0.0.0',8888))
sockfd.listen(5)

#　非阻塞设置
# sockfd.setblocking(False)

# 超时时间
sockfd.settimeout(2)


while True:
    print("Waiting from connect...")
    try:
        connfd,addr = sockfd.accept()
    except (BlockingIOError, timeout) as e:
        sleep(2)
        f.write("%s : %s
"%(ctime(),e))
        f.flush()
    else:
        print("Connect from",addr)
        data = connfd.recv(1024).decode()
        print(data)

I/O多路复用

1. 定义: 同时监控多个IO事件，当哪个IO事件准备就绪就执行哪个IO事件。以此形成可以同时处理多个IO行为，避免一个IO阻塞造成其他IO均无法执行，提高了IO执行效率。

2. 具体方案

 

　　* select方法 ： windows linux unix

　　　　

　　　　rs, ws, xs=select(rlist, wlist, xlist[, timeout])

　　　　功能: 监控IO事件，阻塞等待IO发生

　　　　参数：rlist 列表 存放关注的等待发生的IO事件

　　　　　　　wlist 列表 存放关注的要主动处理的IO事件

　　　　　　    xlist 列表 存放关注的出现异常要处理的IO

　　　　　　　timeout 超时时间　　--> 超时继续执行, 不报错

　　　　返回值： rs 列表 rlist中准备就绪的IO  

　　　　　　　　 ws 列表 wlist中准备就绪的IO

  　　　　　　　  xs 列表 xlist中准备就绪的IO

　　

from socket import *
from select import select

s = socket()
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind(('0.0.0.0',8888))
s.listen(5)

rlist = [s]  
wlist = []
xlist = []

while True:
    rs,ws,xs = select(rlist,wlist,xlist)    # rs列表 --> 里面是rlist中准备就绪的I/O
    for r in rs:
        print(type(r))
        if r is s:
            c,addr = r.accept()
            print("Connect from",addr)
            rlist.append(c) 
        else:
            #　表明有客户端发送消息
            data = r.recv(1024).decode()
            print(data)
            r.send(b'OK')

　

　    * poll方法： linux unix

　　　　　　

　

　　　　p = select.poll()

 

　　　　功能 ： 创建poll对象

　　　　返回值： poll对象

 

　　　　p.register(fd,event)

　　　　功能: 注册关注的IO事件

　　　　参数：fd 要关注的I/O

　　　　　　   event 要关注的IO事件类型

   　　　　　　　　常用类型：POLLIN 读IO事件（rlist）

     　　　　　　　　　　　　 POLLOUT 写IO事件 (wlist)

     　　　　　　　　　　　　 POLLERR 异常IO （xlist）

     　　　　　　　　　　　　 POLLHUP 断开连接

     　　  e.g. p.register(sockfd,POLLIN|POLLERR)

　　　　p.unregister(fd)

　　　　功能：取消对IO的关注

　　　　参数：IO对象或者IO对象的fileno

 

　　　　events = p.poll()

　　　　功能： 阻塞等待监控的IO事件发生

　　　　返回值： 返回发生的IO

　　　　events格式 [(fileno,event),()....]

　　　　每个元组为一个就绪IO，元组第一项是该IO的fileno，第二项为该IO就绪的事件类型

from socket import *
from select import *

#　创建监听套接字，作为关注的ＩＯ
s = socket()
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind(('0.0.0.0',8888))
s.listen(3)

# 创建ｐｏｌｌ对象
p = poll()

#　建立查找字典，通过ＩＯ的fileno查找io对象
#　始终与ｒｅｇｉｓｔｅｒ的ＩＯ保持一直
fdmap = {s.fileno(): s}

#　关注　ｓ
p.register(s,POLLIN|POLLERR)

#　循环监控ＩＯ发生
while True:
    events = p.poll() # 阻塞等待ＩＯ发生
    #　循环遍历查看哪个ＩＯ准备就绪
    for fd,event in events:
        print(events)
        if fd == s.fileno():
            c,addr = fdmap[fd].accept()
            print("Connect from",addr)
            #　关注客户端连接套接字
            p.register(c,POLLIN|POLLHUP)
            fdmap[c.fileno()] = c  #　维护字典
        elif event & POLLIN:
            data = fdmap[fd].recv(1024).decode()
            if not data:
                p.unregister(fd) #　取消监控
                fdmap[fd].close()
                del fdmap[fd] #　从字典删除
                continue
            print(data)
            p.register(fdmap[fd],POLLOUT)
        elif event & POLLOUT:
            fdmap[fd].send(b'OK')
            p.register(fdmap[fd], POLLIN) # 每注册一次都会覆盖上次注册时选定的事件类型

　　

　　　　* epoll方法： linux

 

　　　　　　使用方法 ： 基本与poll相同

　　　　　　　　* 生成对象改为 epoll()

　　　　　　　　* 将所有事件类型改为EPOLL类型

# 创建eｐｏｌｌ对象
ep = epoll()

#　建立查找字典，通过ＩＯ的fileno查找io对象
#　始终与ｒｅｇｉｓｔｅｒ的ＩＯ保持一致
fdmap = {s.fileno():s}

#　关注　ｓ
ep.register(s,EPOLLIN|EPOLLERR)

3. 三种方案的比较

	select	poll	epoll
适用的操作系统		Linux, Unix	Linux
效率	/	低	高
监控I/O数量	/	少	多
触发方式	/	少	多

 

异步I/O

协程

1. 定义：纤程，微线程。是允许在不同入口点不同位置暂停或开始的计算机程序，简单来说，协程就是可以暂停执行的函数。

2. 协程原理 ： 记录一个函数的上下文，协程调度切换时会将记录的上下文保存，在切换回来时进行调取，恢复原有的执行内容，以便从上一次执行位置继续执行。

3. 协程优缺点

 

　　优点

　　　　1. 协程完成多任务占用计算资源很少

　　　　2. 由于协程的多任务切换在应用层完成，因此切换开销少

　　　　3. 协程为单线程程序，无需进行共享资源同步互斥处理

　　缺点

　　　　协程的本质是一个单线程，无法利用计算机多核资源

 

4. Python对协程的支持

　　python3.5以后，使用标准库asyncio和async/await 语法来编写并发代码。asyncio库通过对异步IO行为的支持完成python的协程。

　　虽然官方说asyncio是未来的开发方向，但是由于其生态不够丰富，大量的客户端不支持awaitable需要自己去封装，所以在使用上存在缺陷。更多时候只能使用已有的异步库（asyncio等），功能有限.

import asyncio   #　协程ＩＯ

async def fun1():
    aaaaaaaaaa
    await asyncio.sleep(3)
    bbbbbbbbbb

async def fun2():
    cccccccccc
    await
    ddddddddddd

 　　　　

　　　　* 协程函数的定义需要使用关键字  --> async  def  func()
　　　　* 只有当满足await条件时,才会从fun1中跳出,执行fun2的内容
　　　　* 关于await条件  --> 只支持asyncio库中的内容(如asyncio.sleep()),  而不能支持time.sleep(),  accept(),  recv()等
　　　　* 当我们使用常用的模块,而不是asyncio模块时,无法实现跳转,也就实现不了协程

5. 第三方协程模块

　　* greenlet模块

　　　　安装 ： sudo pip3 install greenlet

 

　　　　函数:

　　　　　　greenlet.greenlet(func)

　　　　　　功能：创建协程对象

　　　　　　参数：协程函数

 

　　　　　　g.switch()

　　　　　　功能：选择要执行的协程函数

from greenlet import greenlet

def fun1():
    print("执行 fun1")
    gr2.switch()
    print("结束 fun1")
    gr2.switch()

def fun2():
    print("执行 fun2")
    gr1.switch()
    print("结束 fun2")

# 将函数变为协程
gr1 = greenlet(fun1)
gr2 = greenlet(fun2)

gr1.switch() # 选择执行的协程函数

　　* gevent模块 

　　　　安装：sudo pip3 install gevent

　　　　函数

 

　　　　　　gevent.spawn(func,argv)

　　　　　　功能: 生成协程对象

　　　　　　参数：func 协程函数

　　　　　　　　   argv 给协程函数传参（不定参）

　　　　　　返回值： 协程对象

　　　　　　gevent.joinall(list,[timeout])

　　　　　　功能: 阻塞等待协程执行完毕  --> 是一个阻塞函数, list中所有的协程函数都执行完毕, 该函数才结束阻塞.

　　　　　　参数：list 协程对象列表

　　　　　　　　   timeout 超时时间

　　　　　　gevent.sleep(sec)

　　　　　　功能: gevent睡眠阻塞

　　　　　　参数：睡眠时间

 

* *************************************************************************************** *

* gevent协程只有在遇到gevent指定的阻塞行为时才会自动在协程之间进行跳转 *

* 如gevent.joinall(),gevent.sleep()带来的阻塞                                                      *

* *************************************************************************************** *

　　　　monkey脚本

　　　　　　作用：在gevent协程中，协程只有遇到gevent指定类型的阻塞才能跳转到其他协程.

　　　　　　　　    因此，我们希望将普通的IO阻塞行为转换为可以触发gevent协程跳转的阻塞，以提高执行效率。

　　　　　　转换方法：gevent 提供了一个脚本程序monkey,可以修改底层解释IO阻塞的行为，将很多普通阻塞转换为gevent阻塞。

　　　　　　使用方法

　　　　　　　　【1】 导入monkey

      　　　　　　　　　　from gevent import monkey

　　　　　　　　【2】 运行相应的脚本，例如转换socket中所有阻塞

      　　　　　　　　　　monkey.patch_socket()

　　　　　　　　【3】 如果将所有可转换的IO阻塞全部转换则运行all

      　　　　　　　　　　monkey.patch_all()

 

　　　　　　　　　　import gevent

　　　　　　　　　   from gevent import monkey

　　　　　　　　　   dir(monkey)

　　　　　　　　　　　　--->  查看有哪些patch_xxx()

 

　　　　　　　　【4】 注意：脚本运行函数需要在对应模块导入前执行

import gevent
from gevent import monkey
monkey.patch_time() # 修改对time模块中阻塞的解释行为
from time import sleep

# 协程函数
def foo(a,b):
    print("Running foo ...",a,b)
    # gevent.sleep(3)
    sleep(3)
    print("Foo again..")

def bar():
    print("Running bar ...")
    # gevent.sleep(2)
    sleep(2)
    print("Bar again..")

# 生成协程对象
f = gevent.spawn(foo,1,2)
g = gevent.spawn(bar)

gevent.joinall([f,g]) #阻塞等待f,g代表的协程执行完毕

协程是单线程, 生成了协程对象后, 只有当线程的后续执行中遇到了gevent阻塞, 协程函数才会执行, 否则协程函数永远不会执行.

gevent遇到阻塞后, 哪个协程函数可以执行就执行哪个协程函数; 上面的代码中, foo函数是先创建的协程函数,遇到joinall阻塞后,先执行的是foo函数.

import gevent
from gevent import monkey
monkey.patch_all() # 执行脚本,修改socket
from socket import *

def handle(c):
    while True:
        data = c.recv(1024).decode()
        if not data:
            break
        print(data)
        c.send(b'OK')
    c.close()

# 创建tcp套接字
s = socket()
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind(('0.0.0.0',8888))
s.listen(5)

# 循环接收来自客户端连接
while True:
    c,addr = s.accept()
    print("Connect from",addr)
    # handle(c) # 处理具体客户端请求
    gevent.spawn(handle,c) # 协程方案