import selectimport socketimport timeclass AsyncTimeoutException(TimeoutError): """ 请求超时异常类 """ def __init__(self, msg): self.msg = msg super(AsyncTimeoutException, self).__init__(msg)class HttpContext(object): """封装请求和相应的基本数据""" def __init__(self, sock, host, port, method, url, data, callback, timeout=5): """ sock: 请求的客户端socket对象 host: 请求的主机名 port: 请求的端口 port: 请求的端口 method: 请求方式 url: 请求的URL data: 请求时请求体中的数据 callback: 请求完成后的回调函数 timeout: 请求的超时时间 """ self.sock = sock self.callback = callback self.host = host self.port = port self.method = method self.url = url self.data = data self.timeout = timeout self.__start_time = time.time() self.__buffer = [] def is_timeout(self): """当前请求是否已经超时""" current_time = time.time() if (self.__start_time + self.timeout) < current_time: return True def fileno(self): """请求sockect对象的文件描述符，用于select监听""" return self.sock.fileno() def write(self, data): """在buffer中写入响应内容""" self.__buffer.append(data) def finish(self, exc=None): """在buffer中写入响应内容完成，执行请求的回调函数""" if not exc: response = b''.join(self.__buffer) self.callback(self, response, exc) else: self.callback(self, None, exc) def send_request_data(self): content = """%s %s HTTP/1.0\r\nHost: %s\r\n\r\n%s""" % ( self.method.upper(), self.url, self.host, self.data,) return content.encode(encoding='utf8')class AsyncRequest(object): def __init__(self): self.fds = [] self.connections = [] def add_request(self, host, port, method, url, data, callback, timeout): """创建一个要请求""" client = socket.socket() client.setblocking(False) try: client.connect((host, port)) except BlockingIOError as e: pass # print('已经向远程发送连接的请求') req = HttpContext(client, host, port, method, url, data, callback, timeout) self.connections.append(req) self.fds.append(req) def check_conn_timeout(self): """检查所有的请求，是否有已经连接超时，如果有则终止""" timeout_list = [] for context in self.connections: if context.is_timeout(): timeout_list.append(context) for context in timeout_list: context.finish(AsyncTimeoutException('请求超时')) self.fds.remove(context) self.connections.remove(context) def running(self): """事件循环，用于检测请求的socket是否已经就绪，从而执行相关操作""" while True: r, w, e = select.select(self.fds, self.connections, self.fds, 0.05) if not self.fds: return for context in r: sock = context.sock while True: try: data = sock.recv(8096) if not data: self.fds.remove(context) context.finish() break else: context.write(data) except BlockingIOError as e: break except TimeoutError as e: self.fds.remove(context) self.connections.remove(context) context.finish(e) break for context in w: # 已经连接成功远程服务器，开始向远程发送请求数据 if context in self.fds: data = context.send_request_data() context.sock.sendall(data) self.connections.remove(context) self.check_conn_timeout()if __name__ == '__main__': def callback_func(context, response, ex): """ :param context: HttpContext对象，内部封装了请求相关信息 :param response: 请求响应内容 :param ex: 是否出现异常（如果有异常则值为异常对象；否则值为None） :return: """ print(context, response, ex) obj = AsyncRequest() url_list = [ {'host': 'www.google.com', 'port': 80, 'method': 'GET', 'url': '/', 'data': '', 'timeout': 5, 'callback': callback_func}, {'host': 'www.baidu.com', 'port': 80, 'method': 'GET', 'url': '/', 'data': '', 'timeout': 5, 'callback': callback_func}, {'host': 'www.bing.com', 'port': 80, 'method': 'GET', 'url': '/', 'data': '', 'timeout': 5, 'callback': callback_func}, ] for item in url_list: print(item) obj.add_request(**item) obj.running() 可能是史上最牛逼的异步IO模块
知识必备： a. socket客户端 obj = socket() # obj.connect((198.1.1.1,80)) obj.connect((http://dig.chouti.com/,80)) # 阻塞 obj.send('GET /index http1.1\r\nhost:...\r\ncontent-type:xxxxx\r\n\r\n') obj.recv(1024) # 最多接收字节 # 阻塞 obj.close() 示例： import socketclient = socket.socket()
# 连接 client.connect(("43.226.160.17",80)) # 阻塞# 发送请求
data = b"GET / HTTP/1.0\r\nhost: dig.chouti.com\r\n\r\n" client.sendall(data) response = client.recv(8096) # 阻塞 print(response)client.close()
总结： 发送Http请求 非阻塞，报错 try 定义一些操作 b. IO多路复用，用来检测【多个】socket对象是否有变化？ socket_list = [] for i in [www.baid.......,.....] client = socket.socket() client.setblocking(False) # 连接 try: client.connect((i,80)) # 连接的请求已经发送出去， except BlockingIOError as e: print(e) socket_list.append(client) # 事件循环 while True: r,w,e = select.select(socket_list,socket_list,[],0.05) # w, 是什么？[sk2,sk3]，连接成功了 for obj in w: obj.send("GET / http/1.0\....") # r，是什么？ [sk2,sk3], 要收数据了 for obj in r: response = obj.recv(...) print(response) 知识点： client.setblocking(False) select.select检测：连接成功，数据回来了 模块： import socket import selectclass Request(object):
def __init__(self,sock,info): self.sock = sock self.info = infodef fileno(self):
return self.sock.fileno() class QinBing(object): def __init__(self): self.sock_list = [] self.conns = []def add_request(self,req_info):
""" 创建请求 req_info: {'host': 'www.baidu.com', 'port': 80, 'path': '/'}, :return: """ sock = socket.socket() sock.setblocking(False) try: sock.connect((req_info['host'],req_info['port'])) except BlockingIOError as e: passobj = Request(sock,req_info)
self.sock_list.append(obj) self.conns.append(obj)def run(self):
""" 开始事件循环,检测：连接成功？数据是否返回？ :return: """ while True: # select.select([socket对象,]) # 可是任何对象，对象一定要fileno方法 # 对象.fileno() # select.select([request对象,]) r,w,e = select.select(self.sock_list,self.conns,[],0.05) # w,是否连接成功 for obj in w: # 检查obj:request对象 # socket, {'host': 'www.baidu.com', 'port': 80, 'path': '/'}, data = "GET %s http/1.1\r\nhost:%s\r\n\r\n" %(obj.info['path'],obj.info['host']) obj.sock.send(data.encode('utf-8')) self.conns.remove(obj) # 数据返回，接收到数据 for obj in r: response = obj.sock.recv(8096) obj.info['callback'](response) self.sock_list.remove(obj) # 所有请求已经返回 if not self.sock_list: break使用： from .qb import QinBing def done1(response): print(response)
def done2(response):
print(response)url_list = [
{'host': 'www.baidu.com', 'port': 80, 'path': '/','callback': done1}, {'host': 'www.cnblogs.com', 'port': 80, 'path': '/index.html','callback': done2}, {'host': 'www.bing.com', 'port': 80, 'path': '/','callback': done2}, ]qinbing = QinBing()
for item in url_list: qinbing.add_request(item)qinbing.run()