IO并发模型

IO 分类

IO分类：阻塞IO ，非阻塞IO，IO多路复用，异步IO等

阻塞IO

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

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

3.阻塞情况：

• 因为某种执行条件没有满足造成的函数阻塞
  e.g. accept input recv

• 处理IO的时间较长产生的阻塞状态
  e.g. 网络传输，大文件读写

非阻塞IO

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

设置套接字为非阻塞IO

sockfd.setblocking(bool)

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

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

sockfd.settimeout(sec)

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

from socket import *
from time import sleep,ctime

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

#　创建套接字
sockfd = socket()
sockfd.bind(('127.0.0.1',9999))
sockfd.listen(3)

# 设置套接字为非阻塞
sockfd.setblocking(False)

#　设置超时检测时间
sockfd.settimeout(3)

while True:
  print("Waiting for connect...")
  try:
    connfd,addr = sockfd.accept()
  except (BlockingIOError,timeout) as e:
    #　如果没有客户端连接，每隔３秒写一个日志
    f.write("%s : %s
"%(ctime(),e))
    f.flush()
    sleep(3)
  else:
    print("Connect from",addr)
    data = connfd.recv(1024).decode()
    print(data)

# Waiting for connect...
# Waiting for connect...
# Waiting for connect...
#log.txt文件逐渐写入
# Thu Jun 27 00:29:29 2019 : timed out
# Thu Jun 27 00:29:35 2019 : timed out
# Thu Jun 27 00:29:41 2019 : timed out

IO多路复用

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

2.具体方案：

• select方法 ： windows linux unix
• poll方法： linux unix
• epoll方法： linux

select 方法

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

select 实现tcp服务

1.将关注的IO放入对应的监控类别列表

2.通过select函数进行监控

3.遍历select返回值列表，确定就绪IO事件

4.处理发生的IO事件

注意：

wlist中如果存在IO事件，则select立即返回给ws

处理IO过程中不要出现死循环占有服务端的情况

IO多路复用消耗资源较少，效率较高

"""
重点代码

思路分析：
1.将关注的IO放入对应的监控类别列表
2.通过select函数进行监控
3.遍历select返回值列表，确定就绪IO事件
4.处理发生的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(3)

#　设置关注列表
rlist = [s]
wlist = []
xlist = [s]

#　循环监控ＩＯ
while True:
  rs,ws,xs = select(rlist,wlist,xlist)
  # 遍历三个返回列表,处理ＩＯ
  for r in rs:
    # 根据遍历到ＩＯ的不同使用ｉｆ分情况处理
    if r is s:
      c,addr = r.accept()
      print("Connect from",addr)
      rlist.append(c) #　增加新的ＩＯ事件
    # else为客户端套接字就绪情况
    else:
      data = r.recv(1024)
      # 客户端退出
      if not data:
        rlist.remove(r) #　从关注列表移除
        r.close()
        continue # 继续处理其他就绪ＩＯ
      print("Receive:",data.decode())
      # r.send(b'OK')
      #　我们希望主动处理这个ＩＯ对象
      wlist.append(r)

  for w in ws:
    w.send(b'OK')
    wlist.remove(w) #　使用后移除

  for x in xs:
    pass

位运算

定义 ： 将整数转换为二进制，按二进制位进行运算

运算符号：

• & 按位与
• | 按位或
• ^ 按位异或
• << 左移
• >> 右移

• 14 --> 01110
• 19 --> 10011
• 14 & 19 = 00010 = 2 一0则0
• 14 | 19 = 11111 = 31 一1则1
• 14 ^ 19 = 11101 = 29 相同为0不同为1
• 14 << 2 = 111000 = 56 向左移动低位补0
• 14 >> 2 = 11 = 3 向右移动去掉低位

poll方法

p = select.poll()

• 功能 ： 创建poll对象
• 返回值： poll对象

p.register(fd,event)   

• 功能: 注册关注的IO事件
• 参数：fd 要关注的IO

event 要关注的IO事件类型

常用类型：

• POLLIN 读IO事件（rlist）
• POLLOUT 写IO事件 (wlist)
• POLLERR 异常IO （xlist）
• POLLHUP 断开连接 

　　p.register(sockfd,POLLIN|POLLERR)

p.unregister(fd)

• 功能：取消对IO的关注
• 参数：IO对象或者IO对象的fileno

events = p.poll()

• 功能： 阻塞等待监控的IO事件发生
• 返回值： 返回发生的IO

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

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

poll_server 步骤：

1. 创建套接字
2. 将套接字register
3. 创建查找字典，并维护
4. 循环监控IO发生
5. 处理发生的IO

"""
尽量掌握

思路分析：
1. 创建套接字作为监控ＩＯ
2. 将套接字register
3. 创建查找字典，并维护(要时刻与注册ＩＯ保持一致)
4. 循环监控IO发生
5. 处理发生的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()

#　建立查找字典，用于通过ｆｉｌｅｎｏ查找ＩＯ对象
fdmap = {s.fileno():s}

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

#　循环监控
while True:
  events = p.poll()
  #　循环遍历发生的事件　ｆｄ-->fileno
  for fd,event in events:
    #　区分事件进行处理
    if fd == s.fileno():
      c,addr = fdmap[fd].accept()
      print("Connect from",addr)
      #　添加新的关注ＩＯ
      p.register(c,POLLIN|POLLERR)
      fdmap[c.fileno()] = c #　维护字典
    #　按位与判定是ＰＯＬＬＩＮ就绪
    elif event & POLLIN:
      data = fdmap[fd].recv(1024)
      if not data:
        p.unregister(fd) #　取消关注
        fdmap[fd].close()
        del fdmap[fd]  #　从字典中删除
        continue
      print("Receive:",data.decode())
      fdmap[fd].send(b'OK')

epoll方法

1.使用方法 ： 基本与poll相同

• 生成对象改为 epoll()
• 将所有事件类型改为EPOLL类型

2.epoll特点：

• epoll 效率比select poll要高
• epoll 监控IO数量比select要多
• epoll 的触发方式比poll要多 （EPOLLET边缘触发）

"""
尽量掌握
"""

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)

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

#　建立查找字典，用于通过ｆｉｌｅｎｏ查找ＩＯ对象
fdmap = {s.fileno():s}

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

#　循环监控
while True:
  events = ep.poll()
  #　循环遍历发生的事件　ｆｄ-->fileno
  for fd,event in events:
    print("亲，你有ＩＯ需要处理哦")
    #　区分事件进行处理
    if fd == s.fileno():
      c,addr = fdmap[fd].accept()
      print("Connect from",addr)
      #　添加新的关注ＩＯ
      #　将触发方式变为边缘触发
      ep.register(c,EPOLLIN|EPOLLERR|EPOLLET)
      fdmap[c.fileno()] = c #　维护字典
    #　按位与判定是EＰＯＬＬＩＮ就绪
    # elif event & EPOLLIN:
    #   data = fdmap[fd].recv(1024)
    #   if not data:
    #     ep.unregister(fd) #　取消关注
    #     fdmap[fd].close()
    #     del fdmap[fd]  #　从字典中删除
    #     continue
    #   print("Receive:",data.decode())
    #   fdmap[fd].send(b'OK')