参考: http://www.cnblogs.com/snailrun/p/3805188.html
异步网络据说能极大的提高网络server的连接速度,所以打算写一个专题,来学习和了解异步网络.因为Python有个非常出名的异步Lib:Twisted,所以就用Python来完成.
OK,首先写一个pythone socket的server段,对开放三个端口:10000,10001,10002.krondo的例子中是每个server绑定一个端口,测试的时候需要分别开3个shell,分别运行.这太麻烦了,就分别用三个Thread来运行这些services.
import optparse import os import socket import time from threading import Thread from io import StringIO txt = '''1111 2222 3333 4444 ''' # 服务端程序处理过程 def server(listen_socket): while True: buf = StringIO(txt) sock, addr = listen_socket.accept() print('Somebody at %s wants poetry!' %(addr,)) while True: try: line = buf.readline().strip() if not line: sock.close() break sock.sendall(line.encode('utf8')) # this is a blocking call print('send bytes to client: %s' %line) except socket.error: sock.close() break time.sleep(0.5) # server每发送一个单词后等待一会 sock.close() print('\n') # 同时开启三个服务端线程,分别在三个端口监听 # 服务端程序为阻塞方式,只能一次服务于一个客户端 def main(): ports = [10000, 10001, 10002] for port in ports: listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) addres = ('127.0.0.1', port) listen_socket.bind(addres) listen_socket.listen(5) print("start listen at: %d" %port) worker = Thread(target = server, args = [listen_socket]) worker.setDaemon(True) worker.start() if __name__ == '__main__': main() while True: time.sleep(0.1) # 如果不sleep的话, CPU会被Python完全占用了
下面是一个client, 用阻塞方式,先后连接这个三个端口的server:
from socket import * # 建立三个客户端,分别连接三个不同的服务端程序, 接收服务端传过来的数据并打印 # 这三个客户端是阻塞方式通信的 if __name__ == '__main__': ports = [10000, 10001, 10002] for port in ports: address = ('127.0.0.1', port) sock = socket(AF_INET, SOCK_STREAM) sock.connect(address) poem = '' while True: data = sock.recv(4) if not data: sock.close() break poem += data.decode('utf8') print(poem) sock.close()
下面用异步的client来读取,代码如下:
import datetime, errno, optparse, select, socket def connect(port): """Connect to the given server and return a non-blocking socket.""" address = ('127.0.0.1', port) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(address) sock.setblocking(0) # 设置为非阻塞模式 return sock def format_address(address): host, port = address return '%s:%s' % (host or '127.0.0.1', port) if __name__ == '__main__': ports = [10000, 10001, 10002] start = datetime.datetime.now() sockets = list(map(connect, ports)) poems = dict.fromkeys(sockets, '') # socket -> accumulated poem sock2task = dict([(s, i + 1) for i, s in enumerate(sockets)]) while sockets: #运用select来确保那些可读取的异步socket可以立即开始读取IO #OS不停的搜索目前可以read的socket,有的话就返回rlist rlist, _, _ = select.select(sockets, [], []) for sock in rlist: data = '' while True: try: new_data = sock.recv(1024) new_data = new_data.decode('utf8') except socket.error as e: if e.args[0] == errno.EWOULDBLOCK: break raise else: if not new_data: break else: print(new_data) data += new_data task_num = sock2task[sock] if not data: print(poems[sock]) # 打印sock接收到的数据 sockets.remove(sock) sock.close() print('Task %d finished\n' % task_num) else: addr_fmt = format_address(sock.getpeername()) msg = 'Task %d: got %d bytes of poetry from %s\n' print(msg % (task_num, len(data), addr_fmt)) poems[sock] += data # 保存每个sock接收到的数据 elapsed = datetime.datetime.now() - start print('Got poems in %s' %elapsed)
结果只需要2秒就完成了读取任务。效率非常明显啊。对客户端的异步改造主要有两点:
- 同步模式下,客户端分别创建socket;而在异步模式下,client开始就创建了所有的socket。
- 通过“sock.setblocking(0)”设置socket为异步模式。
- 通过Unix系统的select来返回可读取socket信息
- 最为核心的是8行和26行。尤其是26行的select操作返回待读取socket的列表。