running executor 服務器 RoCE 進行 break python buffer 創建

ProcessPoolExecutor對multiprocessing進行了高級抽象，暴露出簡單的統一接口。 異步非阻塞 爬蟲 對於異步IO請求的本質則是【非阻塞Socket】+【IO多路復用】：

"""

史上最牛逼的異步IO模塊

"""

import select import socket import time class 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()

python學習筆記之四-多進程&多線程&異步非阻塞