【Python】简单Web框架从零开始（一）：Web服务
一、前言
使用Python可以很方便的搭建一个网站，有很多现成的Web框架可以使用例如：Django，Tornado，web.py，web2py，Flask，Bottle等等。我几年前使用Tornado框架为项目做过一个多人协作的管理平台，后来为了想深入学习就自己写了一个简单的Web框架，并在新项目中投入使用已有1年左右，虽然没有上面提到的那些框架那么强大，但也勉强够用。这次刚好有时间做一个简单的记录。
二、简单Web服务
下图展示了我这次要写的Web框架简单示意图，包含有哪些模块，以及它们的调用层级，模板渲染模块会在后面有空的时候再贴进来。
浏览器向Web服务器发送请求，这个请求使用TCP协议发送到服务器，因此服务器也需要使用TCP协议才能收到请求，创建一个简单的TCP服务器就像下面代码展示的那样。
tcp_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) tcp_server.bind(("0.0.0.0", 80)) tcp_server.listen(5) while True: client, addr = tcp_server.accept() print(u"...接收到连接：", addr) while True: data = client.recv(1024) if not data: break client.send('hello') client.close() tcp_server.close()
打开浏览器输入http://127.0.0.1就会看到hello信息。当然作为Web服务器处理请求需要有一定的效率，上面的代码如果同时有多个请求时，只能逐个进行响应，如果采用I/O多路复用以及引入线程池的话，效果会更好一点。下面的代码引入了采用I/O多路复用的select模块，以及将接收数据投递到线程池中进行处理。这样做可以较好的提升Web服务器的处理性能。
class SimpleHttpServer: def __init__(self, host, port=80): self.http_host = host self.http_port = port self.http_running = False self.http_pool = thread_pool.ThreadExecutor(4) self.sock_selector = selectors.DefaultSelector() self.http_sock = socket.socket() def _setup_server(self): self.http_sock.bind((self.http_host, self.http_port)) self.http_sock.listen(5) self.http_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) ex_data = {"address":(self.http_host, self.http_port), "callback":self._on_new_connected} self.sock_selector.register(self.http_sock, selectors.EVENT_READ, ex_data) return True def _start_forever(self): while self.http_running: try: events = self.sock_selector.select() for obj, mask in events: callback = obj.data["callback"] self.sock_selector.unregister(obj.sock) self.http_pool.submit(callback, obj.sock, obj.data) except KeyboardInterrupt as e: self.http_running = False def _on_new_connected(self, sock, data): client, address = sock.accept() ex_data = {"address":address, "callback":self._on_client_request} self.sock_selector.register(client, selectors.EVENT_READ, ex_data) self.sock_selector.register(sock, selectors.EVENT_READ, data) def _on_client_request(self, sock, data): raw = sock.recv(4096) sock.sendall("hello") sock.close() def start_server(self): if not self._setup_server(): return False self.http_running = True self._start_forever() return True
我用JMeter创建500用户进行并发测试，Throughput大概可以到400/秒，我使用的还是我的10年前的I3双核笔记本进行测试的，开太多用户反而是JMeter太耗了。
三、后记
我的开发环境是Python 2.7 64位，如果使用Python3的可以直接使用自带的selectors和ThreadExecutor模块，另外这次大概会分两到三期写完。最后我会把代码上传到GitHub上面，如果有兴趣欢迎留信息交流。
欢迎微信搜索"游戏测试开发"关注一起沟通交流。
