3.5.4 Selectors(epoll)

Posted 2020-11-24 infinitecodes

Selectors

This module allows high-level and efficient I/O multiplexing, built upon the select module primitives. Users are

该模块允许高级和高效的I/O多路复用，在原select模块的基础上建成。

encouraged to use this module instead, unless they want precise control over the OS-level primitives used.

用户被推荐使用这个（selectors）代替那个（select），除非他们想要对使用的操作系统级原语进行精确控制。

下面是一个selectors的socket示例，可以实现大并发，为了搞清结构和语句，会多次进行print语句查看相关重要信息：

服务器端

import selectors, socket

sel = selectors.DefaultSelector()
#默认用epoll，如果机器上没有epoll，则用select

def accept(sock, mask):    #4只要新链接进来，就调用accept()
    conn, addr = sock.accept()    #5然后链接
    print(‘Connection is: ‘, conn, ‘Address is: ‘, addr, ‘Mask is: ‘, mask)
    conn.setblocking(False)    #6设为非阻塞
    sel.register(conn, selectors.EVENT_READ, read)
    #7将新建立的链接再次注册给sel，新链接只要活动就调用read()函数

def read(conn, mask):
    data = conn.recv(1024)    #9然后开始接收数据
    if data:
        print(‘Echoing: ‘, repr(data))
        conn.send(data)
    else:
        print(‘Close .‘, conn)
        sel.unregister(conn)    #10注销事件
        conn.close()    #11关闭链接

sock = socket.socket()
sock.bind((‘localhost‘, 9999))
sock.listen(500)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)
#1注册事件，把上面写的socket注册给select，只要来新链接，就调用accept函数

while True:
    events = sel.select()
    print(‘33[30;41;1mEvents is: 33[0m‘, events)
    #2这个select是根据系统变化而变化的，支持epoll就用epoll，没有epoll就用select
    #默认阻塞，有活动链接就返回该链接的数据列表
    for key, mask in events:
        callback = key.data    #3data: 返回的events列表中的一项为data，内容就是accept()
        print(‘33[30;43;1mKey Data is: 33[0m‘, key.data,
              ‘33[30;43;1mAccept is: 33[0m‘, accept)
        callback(key.fileobj, mask)    #8fileobj: 同上，就是conn

客户端，我们就用select的客户端

import socket

messages = [‘Saying that I love you‘,
            ‘Is not the words I want to hear from you‘,
            "It‘s not that I want you",
            ‘Not to say, but if you only knew‘,
            ‘How easy it would be to show me how you feel‘,
            ‘More than words is all you have to do to make it real‘,
            "Then you wouldn‘t have to say that you love me",
            "Cause I‘d already know"]
server_address = (‘localhost‘, 9999)

socks = [socket.socket(socket.AF_INET, socket.SOCK_STREAM),
              ]
#客户端发起的socket链接
print(‘Connecting to %s port %s‘ % server_address)
for s in socks:
    s.connect(server_address)
    #开始socks中的每条链接

for m in messages:
#每读取到一条数据
    for s in socks:
    #就让一条socks中的链接
        print(‘%s: sending %s.‘ % (s.getsockname(), m))
        s.send(m.encode(‘utf-8‘))
        #发送这条数据

    for s in socks:
    #同时让每条socks中的链接
        data = s.recv(1024)
        #接收服务器返回的数据
        print(‘%s received data %s.‘ % (s.getsockname(), data))
        if not data:
            print(‘Connection %s has lost.‘ % s.getsockname())
            #没有数据说明链接已断开

1. 导入程序要使用的相关模块，没啥说的；

2. 编写socket，将socket注册给sel实例

3. 重点来了，循环开始了！

执行完events变量赋值后，会自动跳转到客户端，如果你给程序加了断点的话。

加断点的客户端执行到s.connect(server_address)会自动跳转回服务器端

4. 收到链接的服务器端输出链接信息

Events is:  [(SelectorKey(fileobj=<socket.socket fd=756, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=(‘127.0.0.1‘, 9999)>, fd=756, events=1, data=<function accept at 0x0000020492C67510>), 1)]

这里，我们可以看到sel.select()返回的是一个列表，列表中是一个元组，这个元组中总共有2个元素，第一个元素SelectorKey又包含了一个元组，这个元组中的fileobj其实就是接入的socket套接字的详细参数。

5. 第一次循环正式开始，很多人不明白这里面的key, mask, callback, key.data都是什么意思，这里我输出以下，让大家一目了然。

输出

Key is:  SelectorKey(fileobj=<socket.socket fd=792, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=(‘127.0.0.1‘, 9999)>, fd=792, events=1, data=<function accept at 0x000002481DBCC620>) 
Mask is:  1 
Callback is:  <function accept at 0x000002481DBCC620> 
Key Data is:  <function accept at 0x000002481DBCC620> 
Accept is:  <function accept at 0x000002481DBCC620> 
Read is:  <function read at 0x000002481DBCC7B8>

看到了吗，key就是events列表的嵌套元组里的第一个元素，mask是第二个元素，在key中，包含了一个data的参数，这个参数是accept函数名。

6. 然后执行这个函数

这里的callback是什么？就是上面取到的accept，加括号就是调用执行，传入的2个参数，key.fileobj就是接入的客户端的socket，mask暂时不用管他。

7. 执行accept函数内的代码块，此时已建立链接，最后一行将read函数注册给sel示例

输出

Connection is:  <socket.socket fd=1460, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=(‘127.0.0.1‘, 9999), raddr=(‘127.0.0.1‘, 64042)> Address is:  (‘127.0.0.1‘, 64042)

8. 第二次大循环

输出

Events is:  [(SelectorKey(fileobj=<socket.socket fd=1460, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=(‘127.0.0.1‘, 9999), raddr=(‘127.0.0.1‘, 64042)>, fd=1460, events=1, data=<function read at 0x000002481DBCC7B8>), 1)]
Key is:  SelectorKey(fileobj=<socket.socket fd=1460, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=(‘127.0.0.1‘, 9999), raddr=(‘127.0.0.1‘, 64042)>, fd=1460, events=1, data=<function read at 0x000002481DBCC7B8>) 
Mask is:  1 
Callback is:  <function read at 0x000002481DBCC7B8> 
Key Data is:  <function read at 0x000002481DBCC7B8> 
Accept is:  <function accept at 0x000002481DBCC620> 
Read is:  <function read at 0x000002481DBCC7B8>

上一步的最后一行程序做了什么？是不是将read注册给了sel实例，这里callback就变成了read。下一步就执行read函数内的代码块。

9. 这一步没什么好讲的。

没有数据，链接断开，客户端就结束了。

接下来，重头戏。我要展示一下Selectors的魅力之处。当然，我没有linux，只有windows。

将客户端的sock变量那里修改一下，得到这句：

socks = [socket.socket(socket.AF_INET, socket.SOCK_STREAM) for i in range(500)]

同时并发500个链接

结果，就不贴了，自己试一下吧，500个链接，每条链接发8句话，收到8句话，一瞬间完成。很爽，哈哈