6、网络编程六

selectors层次结构
    BaseSelector
    +-- SelectSelector  # 实现select
    +-- PollSelector    # 实现poll
    +-- EpollSelector   # 实现epoll
    +-- DevpollSelector # 实现devpoll
    +-- KqueueSelector  # 实现kqueue

register(fileobj, events, data=None)：注册类文件对象(fileobj)，监控类文件对象的I/O状态，返回与fileobj关联的SelectorKey对象；
unregister(fileobj)：注销一个类文件对象，取消对它的状态监控，它返回与类文件对象关联的SelectorKey对象；
select()：默认永久阻塞，可以设置超时时间，等待注册的fileobje的读或写事件准备好，即开始监控，返回值是一个(key, events)的元组，其中key是一个SelectorKey类的实例， 而events就是event，也就是读或写；
get_map()：返回fileobj对象和其关联的SelectorKey的映射，这个映射是一个字典字典的key为fd，value为SelectorKey，其中SelectorKey就是注册的这个类文件对象的SelectorKey对象，fd就是这个类文件对象的文件描述符；
modify(fileobj, events, data)：用于修改一个注册过的文件对象，比如从监听可读变为监听可写，它其实就是register() 后再跟unregister()，但是使用modify()更高效；
get_key(fileobj)：返回已注册的类文件对象的SelectorKey；

# Choose the best implementation, roughly:
#    epoll|kqueue|devpoll > poll > select.
# select() also can't accept a FD > FD_SETSIZE (usually around 1024)
if 'KqueueSelector' in globals():
    DefaultSelector = KqueueSelector
elif 'EpollSelector' in globals():
    DefaultSelector = EpollSelector
elif 'DevpollSelector' in globals():
    DefaultSelector = DevpollSelector
elif 'PollSelector' in globals():
    DefaultSelector = PollSelector
else:
    DefaultSelector = SelectSelector

class SelectSelector(_BaseSelectorImpl):
    """Select-based selector."""
    def register(self, fileobj, events, data=None): pass
# fileobj：需要监听的类文件对象；
# events：监听的事件，是读事还是写事件；
# data：注册一个类文件对象需要绑定的data，主要用于回调；

EVENT_READ：可读，内核空间的数据已经准备完成，可以开始读了；
EVENT_WIRTE：可写，内核空间的数据已经准备完成，可以开始写了；
# 源码如下
EVENT_READ = (1 << 0)：表示读写状态，读为1；
EVENT_WRITE = (1 << 1)：表示读写状态，写为2；

SelectorKey = namedtuple('SelectorKey', ['fileobj', 'fd', 'events', 'data'])

server=socket.socket()
server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) # 开启端口复用，避免端口已占用
server.bind(("127.0.0.1",8080))
server.listen()
server.setblocking(False) # selectors要求监控对象为非阻塞(经测试，阻塞也可以)

select=selectors.DefaultSelector() # 创建一个select对象
key=select.register(server,events=selectors.EVENT_READ) # 将需要状态监控的类文件对象调用select系统调用，注册到监控器里面去，它返回一个SelectorKey对象
events=select.select() # 开始状态监控，select()方法默认阻塞，直至所监控的所有I/O对象，有一个或者一部分产生了I/O满足events事件
print(events) # 返回已满足事件的I/O对象，一个或者多个
select.close()

server=socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) # 开启端口复用，避免端口已占用
server.bind(("127.0.0.1",8080))
server.listen()
server.setblocking(False) # selectors要求监控对象为非阻塞(经测试，阻塞也可以)

def accept(sock): # 回调函数
    new_sock,client_info=sock.accept()
    print(client_info)

select=selectors.DefaultSelector()
select.register(server,events=selectors.EVENT_READ,data=accept) # data为当server读事件准备好了之后，需要用到的函数，该函数主要用于回调

ready_list=select.select() # 开始状态监控，select()方法默认阻塞，直至所监控的所有I/O对象，有一个或者一部分产生了I/O满足events事件，返回值为产生I/O的类文件对象列表
for key,event in ready_list: # 迭代所有数据准备好I/O对象
    callback=key.data # key.data为registry时传递的函数，其实这就是一个回调函数
    callback(key.fileobj) # key.fileobj就是我们传入监控的socket对象，将key.fileobj传入我们的回调函数，进行回调
select.close()
# ('127.0.0.1', 53342)

server=socket.socket()
server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) # 开启端口复用，避免端口已占用
server.bind(("127.0.0.1",8080))
server.listen()
server.setblocking(False) # selectors要求监控对象为非阻塞(经测试，阻塞也可以)

# 消息收发回调函数
def recv(new_socket):
    data=new_socket.recv(1024)
    if data: # 判断客户端是否已异常断开
        new_socket.send(data)
    else:
        select.unregister(new_socket) # 如果客户端断开，记得将这个new_socket从队列中移除，因为客户端断开服务端会收到一个空串，这个消息如果不收，select就会进入死循环
# 客户端链接回调函数
def accept(sock):
    new_sock,client_info=sock.accept()
    select.register(new_sock,selectors.EVENT_READ,recv) # 将newsocket也加入select队列，使得select能对newsocket得读写事件进行监听

select=selectors.DefaultSelector()
select.register(server,events=selectors.EVENT_READ,data=accept)

while True: # 开启循环，循环监听
    events=select.select() # 开启监听，默认阻塞，直至监控的I/O满足事件，监控第一阶段，返回数据已准备好的所有fileobj
    for key,event in events:  # 迭代所有数据准备好得SelectorKey对象
        key.data(key.fileobj) # 开始进行回调，key.data就是我们registry时传入得回调函数，key.fileobj就是我们传入监控的socket对象

class ChatServer():
    def __init__(self,bind_address="127.0.0.1",bind_port=8083):
        self.bind_address=bind_address,bind_port
        self.server=socket.socket(family=socket.AF_INET,type=socket.SOCK_STREAM)
        self.select=selectors.DefaultSelector()
        self.event=threading.Event() # 创建一个Flag
    def start(self):
        self.server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
        self.server.bind(self.bind_address)
        self.server.listen()
        self.server.setblocking(False) # 设置socket为非阻塞
        self.select.register(self.server,selectors.EVENT_READ,{"callback":self.accept,"recv":False}) # 将accept函数与当前的fileobj进行绑定
        while not self.event.isSet(): # 借助Event来实现死循环
            events=self.select.select() # 使用I/O多路复用监控多路I/O的读事件，默认阻塞，返回一个列表，每个元素为一个二元组，元组内的第一个元素SelectorKey，第二个元素为events
            for key,evnet in events:
                key.data.get("callback")(key.fileobj) # 借助registry时提供的data函数，实现回调
    def accept(self,sock): # 客户端链接回调函数
        new_socket,client_info=sock.accept()
        self.select.register(new_socket,selectors.EVENT_READ,{"callback":self.recv,"recv":True}) # 将accept函数与当前的fileobj进行绑定
    def recv(self,new_socket): # 信息交互回调函数
        data=new_socket.recv(1024)
        if data or data.decode=="q":
            self.group_send(data)
        else: # 垃圾清理
            new_socket.close()
            self.select.unregister(new_socket)
    def group_send(self,data): # 群聊的实现
        for fd,key in self.select.get_map().items(): # 从select的监视器里面拿到所有的与客户端通信的socket
            if key.data.get("recv"): # 判断，如果当前socket为accept的socket就不进行数据发送
                key.fileobj.send(data)
    def stop(self):
        self.event.set()
        for fd,key in self.select.get_map().items():
            key.fileobj.close()
        self.select.close()
        self.server.close()
c1=ChatServer()
c1.start()