Kqueue 实现非阻塞 Socket 通信

如果有误，请大神指出啊！

之前留下的坑

之前写过一篇 kqueue 实现文件操作监控，讲了 Kqueue 在文件监控的应用，文章给出的例子只对于一个 test 文件进行监控。

Kqueue 或者 Epoll 更多的是被使用在 Socket 通信的场景中，于是我又写了一个带有 Socket 的版本。代码如下：

1# coding=utf-8
2import select
3from socket import socket
4from socket import AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR
5from threading import Thread
6
7fd = open('test')
8s = socket(AF_INET, SOCK_STREAM)
9s.bind(("127.0.0.1", 3000))
10s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
11s.listen(1)
12kq = select.kqueue()
13flags = select.KQ_EV_ADD | select.KQ_EV_ENABLE | select.KQ_EV_CLEAR
14fflags = select.KQ_NOTE_DELETE | select.KQ_NOTE_WRITE | select.KQ_NOTE_EXTEND \
15         | select.KQ_NOTE_RENAME
16
17# 监测文件事件，如果有新事件在这个 fd 上发生，则返回，监测事件类型由 fflags 规定
18file_ev = select.kevent(fd.fileno(), filter=select.KQ_FILTER_VNODE, flags=flags, fflags=fflags)
19
20# 监测 Socket 事件，如果有新数据可读则返回
21socket_ev = select.kevent(s.fileno(), filter=select.KQ_FILTER_READ, flags=flags)
22
23# 监测多个对象就只需把很多 kevent 对象塞进 events 列表中，然后传递给 control 函数
24events = []
25events.append(file_ev)
26events.append(socket_ev)
27
28
29# 处理这个 socket 请求
30def socket_handler(cl):
31    while True:
32        data = cl.recv(100)
33        print data
34        if not data:
35            cl.close()
36            print 'socket closed'
37            break
38
39while True:
40    revents = kq.control(events, 1, None)
41    for e in revents:
42        # 如果是 socket 触发的事件
43        if e.ident == s.fileno():
44            print 'Event from socket'
45            if e.filter & select.KQ_FILTER_READ:
46                cl, _ = s.accept()
47                # 如果直接调用 socket_handler 函数，那么这个 eventloop 会被阻塞，所以此处使用线程
48                Thread(None, socket_handler, args=(cl,)).start()
49            else:
50                print e
51        # 如果是文件触发的事件
52        if e.ident == fd.fileno():
53            print 'Event from file'
54            if e.fflags & select.KQ_NOTE_EXTEND:
55                print 'extend'
56            elif e.fflags & select.KQ_NOTE_WRITE:
57                print 'write'
58            elif e.fflags & select.KQ_NOTE_RENAME:
59                print 'rename'
60            elif e.fflags & select.KQ_NOTE_DELETE:
61                print 'delete'
62            else:
63                print e
64

然而随时时间的推移，我发现其实这是有大大的问题的，我自作聪明地为每个新链接产生一个线程应付 Client，然而如果当 Client 源源不断地涌入时，线程数会超标，导致程序发送错误。就算我们维护一个线程的列表，使列表的长度不大于某个标准，每次创建线程的损耗也是存在的。

性能对比

首先我们建立了一个 KqueueEventLoop 的循环类，代码如下：

1# 部分代码参考了 SS 源码中的实现
2class KqueueEventLoop(object):
3
4    KQ_FILTER_READ = select.KQ_FILTER_READ
5
6    def __init__(self):
7        self._fd_map = {}
8        self._handler_map = {}
9        self._event_map = {}
10        self.kq = select.kqueue()
11        self.klist = []
12        self._stop = False
13
14    # 启动这个事件循环
15    def run(self):
16        while not self._stop:
17            events = self.poll()
18            for e in events:
19                self._fd_map[e.ident](self._handler_map[e.ident])
20
21    # 从事件池里面取事件出来
22    def poll(self):
23        events = self.kq.control(self.klist, 1, None)
24        return events
25
26    # 注册事件
27    def add(self, f, mode, handler):
28        fd = f.fileno()
29        event = select.kevent(fd, filter=mode, flags=select.KQ_EV_ADD | select.KQ_EV_ENABLE | select.KQ_EV_CLEAR)
30        self._handler_map[fd] = f
31        self._fd_map[fd] = handler
32        self._event_map[fd] = event
33        self.klist.append(event)
34
35    # 删除事件
36    def remove(self, f):
37        fd = f.fileno()
38        del self._handler_map[fd]
39        del self._fd_map[fd]
40        self.klist.remove(self._event_map[fd])
41
42    # 暂停事件循环
43    def stop(self):
44        self._stop = True

如果是用一开始的代码的想法，为每个 Client 生成一个线程，调用方法如下：

1def test():
2    loop = KqueueEventLoop()
3    s = socket(AF_INET, SOCK_STREAM)
4    s.bind(("127.0.0.1", 3000))
5    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
6    s.listen(5)
7
8    def callback(f):
9        Thread(None, handler, args=(f,)).start()
10
11    def handler(f):
12        print 'INFO: New connection established.'
13        cl, _ = f.accept()
14        while True:
15            data = cl.recv(1024)
16            if not data:
17                print 'INFO: Connection dropped.'
18                loop.remove(cl)
19                cl.close()
20                return
21            print 'DATA: %s' % repr(data)
22
23    loop.add(s, KqueueEventLoop.KQ_FILTER_READ, callback)
24    loop.run()
25
26if __name__ == '__main__':
27    test()

我们模拟了一个 100 个客户端，每隔 0.1 秒向服务器发 1，获取到该服务端内存使用量为 14MB。之前提到这边线程的产生会造成两个严重的问题，所以我们就不生成线程。那么怎么做到原来的 while 循环来接受这么多客户端发来的数据呢？答案是，不用 while 循环。

继续注册事件

当事件池返回一个新连接建立请求时，接受并建立这个连接，再把这个连接扔回事件池中，代码如下：

1def test():
2    loop = KqueueEventLoop()
3    s = socket(AF_INET, SOCK_STREAM)
4    s.bind(("127.0.0.1", 3000))
5    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
6    s.listen(5)
7
8    def handler(f):
9        print 'INFO: New connection established.'
10        cl, _ = f.accept()
11        cl.setblocking(False)
12        loop.add(cl, KqueueEventLoop.KQ_FILTER_READ, read_data)
13
14    def read_data(cl):
15        try:
16            data = cl.recv(1024)
17            if not data:
18                print 'INFO: Connection dropped.'
19                loop.remove(cl)
20                cl.close()
21                return
22            print 'DATA: %s' % repr(data)
23        except Exception, e:
24            print 'ERROR: %s' % repr(e)
25            loop.remove(cl)
26            cl.close()
27
28    loop.add(s, KqueueEventLoop.KQ_FILTER_READ, handler)
29    loop.run()
30
31if __name__ == '__main__':
32    test()

我们通过 loop.add(cl, KqueueEventLoop.KQ_FILTER_READ, read_data) 将每个新连接 cl 加入到事件循环 loop 中，并让事件循环类记录回调方法 read_data。此时的状况就是，整个程序只有一个事件循环，每来一个新连接，通过 handler 回调建立该连接。再如果网卡接受到新数据，KqueueEventLoop 找到新数据对应的连接，把数据读入内存并打印出来。

以上实现没有了新建线程带来的损耗，整个程序只保留了一个 while 循环，使得相同情况下内存使用量仅为 5MB。并且并发连接数量有很大提升，不再受限于线程数限制。

其他

在 IO 密集型程序中，其实线程的使用是可以改善程序运行速度的。但线程还是要用在可控的地方，比如用线程去跑事件循环。

Kqueue 实现非阻塞 Socket 通信

之前留下的坑

性能对比

继续注册事件

其他

评论区

相关推荐