IO多路复用

写在最前面：

一.网络编程里有两类socket必须弄清楚，第一类socket是专门用来接收用户的连接请求，另一类socket是专门用来发送和接收数据的

1.server_sock = socket.socket() # 专门用来接收用户的连接请求
2.sock.addr = server_sock.accept() # 如果客户端没有发送连接请求，会一直阻塞 ；如果有新的客户端来连接服务器，那么就产生了一个新的套接字专门为发送和接收数据
          sock.recv()
          sock.send()

二.accept是发生在三次握手之后，具体过程参考:
1.牛客网
2.Socket 之accept与三次握手的关系
三.客户端调用connect，内部只有当三次握手完成之后，才会从阻塞变成非阻塞
四.整个服务端响应过程

1.sersocket = socket().socket() # 创建一个socket
2.sersocket.bind(('0.0.0.0',8000)) # 绑定ip和端口号
3.sersocket.listen(num) # 监听连接请求，内部维护了一个队列，存放着每一个完成TCP三次握手后的连接请求信息
4. sock, addr = sersocket.accept() # 从队列中获取连接请求，并封装成socket返回
5.sock.send(data) # 发送数据给客户端，非阻塞，数据拷贝到发送缓存区就返回
6.sock.recv(1024) # 接收数据，过程是阻塞式的

简单的server端

import socket

server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
print(server) 

server.bind(('0.0.0.0',8000))
server.listen() # 监听多个连接请求

while True:
    sock, addr = server.accept()  # 如果客户端没有发送连接请求，会一直阻塞 
    print(sock,addr)
    data = sock.recv(1024)
    print(data.decode('utf-8'))
    sock.send(b'hello world')

注意：
server socket对象：

<socket.socket fd=3, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('0.0.0.0', 0)>

sock对象带有客户端的连接信息：

<socket.socket fd=4, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8000), raddr=('127.0.0.1', 60870)> ('127.0.0.1', 60870)

上面的过程可以简化成

image

IO多路复用

I/O多路复用也就是很多网络连接(多路)，共(复)用一个线程

I/O多路复用就通过一种机制，可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。但select，poll，epoll本质上都是同步I/O，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需自己负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。IO多路复用的技术是由操作系统提供的功能。

利用select实现IO多路复用

select是操作系统为我们提供解决IO多路复用的机制，调用select函数会阻塞当前进程，直到维护的socket列表中(该列表中会有连接请求的socket，也会有发送和接收数据的socket)的socket变成可读或可写时，函数才会返回并往下继续执行。

image.png

import socket
import select

sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.bind(('0.0.0.0',8000))
sk.setblocking(False)
sk.listen()

read_lst = [sk]
while True: # [sk,coon]
    r_lst,w_lst,x_lst = select.select(read_lst,[],[])
    for i in r_lst:
        if i is sk: # 只有sk有accept方法，其他socket只能接收和发送数据
            conn,addr = i.accept()
            read_lst.append(conn)
        else:
            ret = i.recv(1024)
            if ret == b'': # 如果发送的是空数据，就断开连接
                i.close()
                read_lst.remove(i)
                continue
            print(ret)
            i.send(b'HTTP/1.1 200 OK \n\n hello')

select的缺点：

• 操作系统轮询socket的状态会降低效率
• 轮询的socket数量有限
• 如果轮询到第三个socket时，第二个socket变成了可读，不得不错过

利用epoll实现IO多路复用

epoll：不再让操作系统轮询socket的状态，而是对每一个socket都绑定了回调函数，一但有socket变成可读或者可写的事件触发，将会调用回调函数。

from socket import *
import selectors

sel=selectors.DefaultSelector() # 该模块会自己选择支持操作系统的select,poll或者epoll

# 用于绑定连接请求socket的回调函数
def accept(sk,mask):
    conn,addr=sk.accept()
    sel.register(conn,selectors.EVENT_READ,data = read)

# 用于绑定非连接请求socket的回调函数
def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print('closing',conn)
            sel.unregister(conn)
            conn.close()
            return
        print(data)
        conn.send(data.upper()+b'_SB')
    except Exception:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()

sk = socket(AF_INET,SOCK_STREAM)
sk.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
sk.bind(('0.0.0.0',8080))
sk.listen(5)
sk.setblocking(False) #设置socket的接口为非阻塞
sel.register(sk,selectors.EVENT_READ,data = accept) #相当于往select的读列表里append了一个sk,并且绑定了一个回调函数accept

while True:  # 这个while就是事件循环
    events=sel.select() #检测所有的socket状态，是否有完成wait data的，阻塞ing
    for sel_obj,mask in events:
        callback=sel_obj.data # callback=accpet
        callback(sel_obj.fileobj,mask) # accpet(sk,1)

客户端

from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8080))

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

为什么IO多路复用可以处理高并发问题呢？

我觉得最主要的是充分利用了服务端接收数据的时间（socket.recv），由于recv是阻塞式的，只有当数据到达内核之后才可以去read。在这段时间内，服务器完全可以尽可能的多接收其他用户的连接请求，而不必阻塞着等着read数据，只有当数据准备好了，我们再去read。

listen参数问题？

之前一直有一个问题困扰着我，就是如果我的服务器程序正在处理耗时任务，突然有新的连接过来请求，这个请求是谁来保存的，毕竟此时我还没有回去执行accpet。其实listen会帮我们维护一个队列，队列的长度就是listen的参数（linux无论设置多少，都是由操作系统决定的），当有新的请求过来的时候，操作系统里的协议栈会帮我们完成三次握手，然后将连接信息放到队列中去，当我们accpet的时候，会从这个队列中取出连接信息生成新的socket用于通信。

我们可以在Mac上做如下实验

# server.py
from socket import *
from time import sleep

tcpsersocket = socket(AF_INET,SOCK_STREAM)

tcpsersocket.bind(('0.0.0.0',8003))

tcpsersocket.listen(5)

while True:
    newsocket,addr = tcpsersocket.accept()
    print(addr)
    sleep(10)

# client.py
from socket import *

for i in range(10):
    c = socket(AF_INET, SOCK_STREAM)
    c.connect(('127.0.0.1',8004))
    print(i) # 一开始会连续打印，当到了5之后开始，之后会每隔10s打印

参考：

1.http://www.cnblogs.com/Eva-J/articles/8324837.html
2.https://www.yanxurui.cc/posts/linux/2017-06-03-IO-multiplexing/
3.https://www.jianshu.com/p/dfd940e7fca2
4.https://blog.csdn.net/jiange_zh/article/details/50811553