自己实现多线程的socket，socketserver源码剖析

1，IO多路复用

三种多路复用的机制：select、poll、epoll

用的多的两个：select和epoll

简单的说就是：
1，select和poll所有平台都支持，epoll只有linux支持
2，select效率不高，epoll效率高
3，IO多路复用用来监听socket对象内部是否变化
4，要调用select模块

 

什么是文件描述符：

当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。文件描述符的有效范围是 0 到 OPEN_MAX

linux查看OPEN_MAX的方法:ulimit -n。linux默认最大文件描述符为1024，一般可以将它改为65535，可以使用命令ulimit -HSn  65535。也可以保存到/etc/security/ulimit.conf里

 

 

 

 

python的select函数：

 

def select(rlist, wlist, elist, timeout=None)->(rl,wl,el)

rlist获取变化的句柄添加到rl中

只要wlist有句柄，都放到wl中

当elist某个句柄发生错误时，放到el中

timeout，如果没有设置超时时间，select一直会卡着，如果设置timeout=1，表示句柄没有变化时，select会卡1秒，一有变化就执行

 

在socket中

select用来监听sk连接时候的句柄和收发数据的句柄

import socket
import select
sk=socket.socket()
sk.bind(("127.0.0.1",9999))
sk.listen(2)
inputs=[sk]
outputs=[]
msg={}
while True:
    rlist,wlist,e=select.select(inputs,outputs,[],1)
print(len(inputs),len(rlist),len(wlist),len(outputs))
'''
    只要有连接进来，就接收，只要有收发消息的句柄发生变化，就收消息，
    收到消息后可以不直接发送，存到outputs里，wlist==outputs，
    只要outputs里有句柄，就交给wlist去循环发送
'''
for r in rlist:
if r==sk:
            conn,addr=r.accept()
            inputs.append(conn)
else:
try:
                rt=r.recv(1024)
                outputs.append(r)
print(rt)
                msg.setdefault(r, [])
                msg[r].append(rt)
except:
                inputs.remove(r)
del msg[r]
#上面如果没有数据，这边就一直发数据给客户端，直到客户端接收消息，得到的是一大串
for w in wlist:
for m in msg[w]:
            w.sendall(m)
        outputs.remove(w)     #所以发送完这边要删除句柄，解决了无限发数据的问题

 

socketserver源码剖析

通过上面的实验得出规律：一开始建立socket对象到listen这几步都没变，直到accept之间在循环使用select检查sk是否变化，如果有新链接进来就accept，之后就能通信了。

import socketserver

class MyServer(socketserver.BaseRequestHandler):

def handle(self):
# print self.request,self.client_address,self.server
        conn = self.request
while True:
            recv_data=conn.recv(5)

            conn.sendall(recv_data)


if __name__ == '__main__':
    server = socketserver.ThreadingTCPServer(('127.0.0.1',8009),MyServer)
print(server.server_address)
    server.serve_forever()

第一步：建立socket对象到listen这几步肯定在创建构造函数的时候已经做掉了。

第二步：因为它是多线程，select查看有sk变动后，每次有连接进来就分配一个线程给sk，然后accept连接

第三步：在accept之后调用socketserver.BaseRequestHandler来收发消息，在调用MySocket的handle，也就是调用BaseRequestHandler的handle

第四步：forever()

ThreadingTCPServer的继承关系:

 

 

 

import socket
import select
import threading

def handle(sk):
'''此处发送'''
    sk.sendall(bytes("hello",encoding="utf-8"))


sk=socket.socket()
sk.bind(("127.0.0.1",9999))
sk.listen(5)
while True:
    rlist,w,e=select.select([sk],[],[],1)
for r in rlist:
        conn,addr=r.accept()
        th=threading.Thread(target=handle,args=(conn,))
        th.daemon=False
        th.start()
sk.close()

import socket

sk=socket.socket()
sk.connect(("127.0.0.1",9999))
print(sk.recv(1024))
sk.close()