本文共 4826 字，大约阅读时间需要 16 分钟。

服务端异步IO配合协程浅析

代码如下

#coding:utf-8import socketfrom selectors import DefaultSelector, EVENT_READ, EVENT_WRITEselector = DefaultSelector()stopped = Falseresponse = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n'response += b'hello world'class Future:    def __init__(self):        self.result = None        self._callback = []    def add_done_callback(self, fn):        self._callback.append(fn)    def set_result(self, result):        self.result = result        for fn in self._callback:            fn(self)    def __iter__(self):        yield self        return self.resultclass Task:    def __init__(self, coro):        self.coro = coro        f = Future()        f.set_result(None)        self.step(f)    def step(self, future):        try:            next_future = self.coro.send(future.result)            print("next_future: ", next_future)        except StopIteration as e:            print("step stop :", e.value)            return        next_future.add_done_callback(self.step)class ConHandler(object):    def __init__(self, con, addr):        self.con = con                                      # 新连接实例        self.addr = addr                                    # 新连接地址    def _read_ready(self):        f = Future()        def sock_read():                                    # 由于没有解析相应的流协议，所以这里直接接受所有的数据，                                                            # 如果解析具体协议也可以改写成协程            all_data = b""                                  # 接受数据            while True:                                     # 一直接受数据直到数据接受完成                try:                    data = self.con.recv(10)                    if data:                        print("recv data : ", data)                        all_data += data                    else:                        print("break while")                        break                except BlockingIOError:                    print("BlockingIOError")                    break            print("all_data", all_data)            # selector.unregister(self.con.fileno())            f.set_result(None)                              # 当数据接受完成后，调用f实例的回调方法，进入将主流程往下执行一步        selector.register(self.con.fileno(), EVENT_READ, sock_read)     # 注册连接的读事件并注册回调函数        yield f        selector.unregister(self.con.fileno())              # 取消连接的注册事件    def fetch(self):        # 读事件到来一次读完        fu = yield from self._read_ready()                  # 调用读数据        print("recv_data: ", fu)        # handler  处理请求        self.response = response*10000        result = yield from self.sock_send_all()            # 将处理的数据发送出去        print("send after : ", result)    def sock_send_all(self):        send_length = yield from self.sock_send()           # 发送相应数据，返回发送的数据长度        self.response = self.response[send_length:]         # 减去已经发送的数据长度        while send_length:                                  # 如果发送长度不为0            send_length = yield from self.sock_send()       # 继续发送数据            self.response = self.response[send_length:]     # 减去已经发送的数据长度        return    def sock_send(self):        f = Future()        def _sock_send():            con_length = self.con.send(self.response)            f.set_result(con_length)        selector.register(self.con.fileno(), EVENT_WRITE, _sock_send)        send_length = yield f        selector.unregister(self.con.fileno())        return send_lengthdef create_server():    sock = socket.socket()                                      # 创建连接    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 设置端口重用    sock.bind(('0.0.0.0', 9003))                                # 设置监听连接端口    sock.listen(10)                                             # 设置监听缓存区    sock.setblocking(0)                                         # 设置连接非阻塞    def sock_accept():                                          # 如果sock的新请求连接处理回调函数        try:            conn, addr = sock.accept()                          # 接受新请求            conn.setblocking(False)                             # 设置接受新连接非阻塞            print(conn)        except Exception:            pass        Task(ConHandler(conn, addr).fetch())                # 用任务类包裹处理流程，使之流程协程处理    selector.register(sock.fileno(), EVENT_READ, sock_accept)   # 注册sock的连接读事件并设置读事件的回调函数def loop():    while not stopped:        events = selector.select()                              # 获取触发的事件        for event_key, event_mask in events:            callback = event_key.data                           # 获取触发事件的回调函数            callback()                                          # 执行回调函数if __name__ == '__main__':    create_server()    loop()

基于上一篇的异步协程爬虫，现在服务端的大概原理是:

转载地址：https://blog.csdn.net/qq_33339479/article/details/78599186 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！