Python Select模型

Posted on

IO多路复用

IO多路复用就是我们经常说的select epoll.select和epoll的好处是单个process就可以同时处理多个网络IO。基本原理是select\epoll会不断的轮询所负责的所有socket，当有某个socket数据到达了，就通知用户进程。
下面是流程图：

当用户进程调用了select，那么整个进程会被block，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel拷贝到用户进程。

注意1：select函数返回结果中如果有文件可读了，那么进程就可以通过调用accept()或recv()来让kernel将位于内核中准备到的数据copy到用户区。

注意2: select的优势在于可以处理多个连接，不适用于单个连接

selectors

基于select模块实现的IO多路复用

`

IO多路复用实现机制

在不同的平台上是不一样的，win平台只有select,Linux平台有select poll epoll

  • win： select

  • linux : select poll epoll

通常是用户空间创建fd,然后copy到内核空间

如果是开fd的数量多，select的的效率低

基于select模块实现的IO多路复用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import selectors
import socket
sock = socket.socket()
sock.bind(("127.0.0.1", 8810))
sock.listen(5)  # 这里虽然设置了最大连接数，但是已经没有限制了
sel = selectors.DefaultSelector()  # 实例化一个对象，会根据不同的平台自动设置优先级
# epoll|kqueue|devpoll > poll > select.  所以Linux系统会自动设置成epoll  win 自动设置成select
# 第二步
def read(conn, mask):
    # pass
    try:  # win 检测异常  当有异常 如客户端断开的时候 
        data = conn.recv(1024)
        print(data)
        print(data.decode("utf-8"))
        data2 = input(">>>")
        conn.send(data2.encode("utf-8"))
    except Exception:
        sel.unregister(conn)  # 解除注册
# 第一步
def accept(sock, mask):  # mask 是没有用的
    conn, addr = sock.accept()
    # print(conn)
    sel.register(conn, selectors.EVENT_READ, read)  # 把conn 添加到列表中
# 首先要注册 只是把sock和 accept绑定
sel.register(sock, selectors.EVENT_READ, accept)  # 注册，但是没有监听accept函数
# 监听
while 1:
    print("waiting...")
    # event 就是那个r
    events = sel.select()  # [(key,mask) ,(key,mask) ,(key,mask) ,)]  # 活活动的对象会自动添加到这里
    for key, mask in events:  # events 是个列表 需要遍历
        print(key.data)  # 拿到accept函数
        print(key.fileobj)  # 当前的活动的对象 sock 文件句柄
        func = key.data  # 调用
        obj = key.fileobj  #
        func(obj, mask)  # 第一个参数是sock 对象
        # break

select缺点：

  • 每次调用slect都要将所有的fd拷贝到内核空间（每次都要拷贝），导致效率下降
  • 监听的的实现是通过遍历所有的fd,（遍历消耗的时间消耗多）判断是否有数据访问
  • 最大连接数（input中放的文件描述符数量1024）

poll：

最大连接数没有限制了，除此之外，和select一样，所以基本不用

epoll:

内部通过3个函数实现（select是一个）

  • 第一个函数：
    创建epoll句柄，把所有的fd拷贝到内核空间，只需要拷贝一次

  • 第二个函数： 回掉函数
    某一个函数或者动作成功完成后，会自动触发一个函数
    为所有的fd绑定一个回调函数,一旦有数据访问，触发改回调函数,回调函数把fd放到链表中。（只要有活动，把fd放到链表中，动态监听）这样就提高了效率
    例子：交试卷

  • 第三个函数，判断链表是否为空

selectors.DefaultSelector()
selectors会根据自己的平台选择最佳IO多路复用，自动选择。win只有select
linux的中epoll中的优先级最高

队列queue

和线程有关系的，在多线程的时候有用，保证信息安全的

队列是一种数据类型

优点：
保证线程安全，不用自己加锁

put get

先进先出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import queue
q = queue.Queue(3) # 默认是先进先出  FIFO  设置参数是最大的存放数量5
q.put(111)
q.put("hello")
q.put(222)
q.put(333,False)  # 默认blocking = True ,False 是当存满的时候，自动报错，解除阻塞的状态
print(q.get())
print(q.get())
print(q.get())
print(q.get()) # 第4次已经拿不到了 取不到 默认阻塞
q.get(False) # 解除阻塞状态

先进后出

1
2
3
4
5
q = queue.LifoQueue()
q.put(111)
q.put(222)
print(q.get())
print(q.get())

优先级

1
2
3
4
5
6
7
8
9
q = queue.PriorityQueue()
q.put([4,"hello4"])
q.put([1,"hello1"])
q.put([2,"hello2"])
print(q.get())
print(q.get())
print(q.get())

join 与task_done方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import queue
q = queue.Queue(5)
q.put(111)
q.put(222)
q.get()
q.task_done()  #
q.get()
q.task_done()  #
q.join()  # 等待task_done  和events是一个原理
print('endnig')

join 与task_done方法必须配合使用

其他的用法

1
2
3
4
5
6
7
8
9
10
q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get([block[, timeout]]) 获取队列，timeout等待时间
q.get_nowait() 相当q.get(False)非阻塞 
q.put(item) 写入队列，timeout等待时间
q.put_nowait(item) 相当q.put(item, False)
q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作

生产者消费者模型

有生产数据的线程
有消费数据的线程

通过一个容器来解决生产者消费者强耦合的问题
这个容器是用来解耦的（类似吃饭的时候的服务员）
目录结构也是一种解耦

下面是用队列模拟实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
  count = 0
  while count <10:
    print("making........")
    time.sleep(random.randrange(3))
    q.put(count)
    print('Producer %s has produced %s baozi..' %(name, count))
    count +=1
    #q.task_done()
    #q.join()
    print("ok......")
def Consumer(name):
  count = 0
  while count <10:
    time.sleep(random.randrange(4))
    if not q.empty():
        data = q.get()
        #q.task_done()
        #q.join()
        print(data)
        print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))
    else:
        print("-----no baozi anymore----")
    count +=1
p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))
# c2 = threading.Thread(target=Consumer, args=('C',))
# c3 = threading.Thread(target=Consumer, args=('D',))
p1.start()
c1.start()
# c2.start()
# c3.start()

© 2018 Peter's Blog Center All Rights Reserved.
Theme by hiero