高性能服務(wù)器Tornado
Python的web框架名目繁多，各有千秋。正如光榮屬于希臘，偉大屬于羅馬。Python的優(yōu)雅結(jié)合WSGI的設(shè)計，讓web框架接口實現(xiàn)千秋一統(tǒng)。WSGI 把應(yīng)用（Application）和服務(wù)器（Server）結(jié)合起來。Django 和 Flask 都可以結(jié)合 gunicon 搭建部署應(yīng)用。
與 django 和 flask 不一樣，tornado 既可以是 wsgi 應(yīng)用，也可以是 wsgi 服務(wù)。當(dāng)然，選擇tornado更多的考量源于其單進(jìn)程單線程異步IO的網(wǎng)絡(luò)模式。高性能往往吸引人，可是有不少朋友使用之后會提出疑問，tornado號稱高性能，實際使用的時候卻怎么感受不到呢？
實際上，高性能源于Tornado基于Epoll（unix為kqueue）的異步網(wǎng)絡(luò)IO。因為tornado的單線程機(jī)制，一不小心就容易寫出阻塞服務(wù)（block）的代碼。不但沒有性能提高，反而會讓性能急劇下降。因此，探索tornado的異步使用方式很有必要。
Tornado 異步使用方式
簡而言之，Tornado的異步包括兩個方面，異步服務(wù)端和異步客戶端。無論服務(wù)端和客戶端，具體的異步模型又可以分為回調(diào)（callback）和協(xié)程（coroutine）。具體應(yīng)用場景，也沒有很明確的界限。往往一個請求服務(wù)里還包含對別的服務(wù)的客戶端異步請求。
服務(wù)端異步方式
服務(wù)端異步，可以理解為一個tornado請求之內(nèi)，需要做一個耗時的任務(wù)。直接寫在業(yè)務(wù)邏輯里可能會block整個服務(wù)。因此可以把這個任務(wù)放到異步處理，實現(xiàn)異步的方式就有兩種，一種是yield掛起函數(shù)，另外一種就是使用類線程池的方式。請看一個同步例子：
|
1
2
3
4
5
6
|
class SyncHandler(tornado.web.RequestHandler): def get(self, *args, **kwargs): # 耗時的代碼 os.system("ping -c 2 www.google.com") self.finish('It works') |
使用ab測試一下：
|
1
|
ab -c 5 -n 5 http://127.0.0.1:5000/sync |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
Server Software: TornadoServer/4.3Server Hostname: 127.0.0.1Server Port: 5000Document Path: /syncDocument Length: 5 bytesConcurrency Level: 5Time taken for tests: 5.076 secondsComplete requests: 5Failed requests: 0Total transferred: 985 bytesHTML transferred: 25 bytesRequests per second: 0.99 [#/sec] (mean)Time per request: 5076.015 [ms] (mean)Time per request: 1015.203 [ms] (mean, across all concurrent requests)Transfer rate: 0.19 [Kbytes/sec] received |
qps 僅有可憐的 0.99，姑且當(dāng)成每秒處理一個請求吧。
下面祭出異步大法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
class AsyncHandler(tornado.web.RequestHandler): @tornado.web.asynchronous @tornado.gen.coroutine def get(self, *args, **kwargs): tornado.ioloop.IOLoop.instance().add_timeout(1, callback=functools.partial(self.ping, 'www.google.com')) # do something others self.finish('It works') @tornado.gen.coroutine def ping(self, url): os.system("ping -c 2 {}".format(url)) return 'after' |
盡管在執(zhí)行異步任務(wù)的時候選擇了timeout 1秒，主線程的返回還是很快的。ab壓測如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
Document Path: /asyncDocument Length: 5 bytesConcurrency Level: 5Time taken for tests: 0.009 secondsComplete requests: 5Failed requests: 0Total transferred: 985 bytesHTML transferred: 25 bytesRequests per second: 556.92 [#/sec] (mean)Time per request: 8.978 [ms] (mean)Time per request: 1.796 [ms] (mean, across all concurrent requests)Transfer rate: 107.14 [Kbytes/sec] received |
上述的使用方式，通過tornado的IO循環(huán)，把可以把耗時的任務(wù)放到后臺異步計算，請求可以接著做別的計算。可是，經(jīng)常有一些耗時的任務(wù)完成之后，我們需要其計算的結(jié)果。此時這種方式就不行了。車道山前必有路，只需要切換一異步方式即可。下面使用協(xié)程來改寫：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
class AsyncTaskHandler(tornado.web.RequestHandler): @tornado.web.asynchronous @tornado.gen.coroutine def get(self, *args, **kwargs): # yield 結(jié)果 response = yield tornado.gen.Task(self.ping, ' www.google.com') print 'response', response self.finish('hello') @tornado.gen.coroutine def ping(self, url): os.system("ping -c 2 {}".format(url)) return 'after' |
可以看到異步在處理，而結(jié)果值也被返回了。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
Server Software: TornadoServer/4.3Server Hostname: 127.0.0.1Server Port: 5000Document Path: /async/taskDocument Length: 5 bytesConcurrency Level: 5Time taken for tests: 0.049 secondsComplete requests: 5Failed requests: 0Total transferred: 985 bytesHTML transferred: 25 bytesRequests per second: 101.39 [#/sec] (mean)Time per request: 49.314 [ms] (mean)Time per request: 9.863 [ms] (mean, across all concurrent requests)Transfer rate: 19.51 [Kbytes/sec] received |
qps提升還是很明顯的。有時候這種協(xié)程處理，未必就比同步快。在并發(fā)量很小的情況下，IO本身拉開的差距并不大。甚至協(xié)程和同步性能差不多。例如你跟博爾特跑100米肯定輸給他，可是如果跟他跑2米，鹿死誰手還未定呢。
yield掛起函數(shù)協(xié)程，盡管沒有block主線程，因為需要處理返回值，掛起到響應(yīng)執(zhí)行還是有時間等待，相對于單個請求而言。另外一種使用異步和協(xié)程的方式就是在主線程之外，使用線程池，線程池依賴于futures。Python2需要額外安裝。
下面使用線程池的方式修改為異步處理：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from concurrent.futures import ThreadPoolExecutorclass FutureHandler(tornado.web.RequestHandler): executor = ThreadPoolExecutor(10) @tornado.web.asynchronous @tornado.gen.coroutine def get(self, *args, **kwargs): url = 'www.google.com' tornado.ioloop.IOLoop.instance().add_callback(functools.partial(self.ping, url)) self.finish('It works') @tornado.concurrent.run_on_executor def ping(self, url): os.system("ping -c 2 {}".format(url)) |
再運(yùn)行ab測試：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
Document Path: /futureDocument Length: 5 bytesConcurrency Level: 5Time taken for tests: 0.003 secondsComplete requests: 5Failed requests: 0Total transferred: 995 bytesHTML transferred: 25 bytesRequests per second: 1912.78 [#/sec] (mean)Time per request: 2.614 [ms] (mean)Time per request: 0.523 [ms] (mean, across all concurrent requests)Transfer rate: 371.72 [Kbytes/sec] received |
qps瞬間達(dá)到了1912.78。同時，可以看到服務(wù)器的log還在不停的輸出ping的結(jié)果。
想要返回值也很容易。再切換一下使用方式接口。使用tornado的gen模塊下的with_timeout功能（這個功能必須在tornado>3.2的版本）。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
class Executor(ThreadPoolExecutor): _instance = None def __new__(cls, *args, **kwargs): if not getattr(cls, '_instance', None): cls._instance = ThreadPoolExecutor(max_workers=10) return cls._instanceclass FutureResponseHandler(tornado.web.RequestHandler): executor = Executor() @tornado.web.asynchronous @tornado.gen.coroutine def get(self, *args, **kwargs): future = Executor().submit(self.ping, 'www.google.com') response = yield tornado.gen.with_timeout(datetime.timedelta(10), future, quiet_exceptions=tornado.gen.TimeoutError) if response: print 'response', response.result() @tornado.concurrent.run_on_executor def ping(self, url): os.system("ping -c 1 {}".format(url)) return 'after' |
線程池的方式也可以通過使用tornado的yield把函數(shù)掛起，實現(xiàn)了協(xié)程處理。可以得出耗時任務(wù)的result，同時不會block住主線程。
|
1
2
3
4
5
6
7
8
9
10
|
Concurrency Level: 5Time taken for tests: 0.043 secondsComplete requests: 5Failed requests: 0Total transferred: 960 bytesHTML transferred: 0 bytesRequests per second: 116.38 [#/sec] (mean)Time per request: 42.961 [ms] (mean)Time per request: 8.592 [ms] (mean, across all concurrent requests)Transfer rate: 21.82 [Kbytes/sec] received |
qps為116，使用yield協(xié)程的方式，僅為非reponse的十分之一左右。看起來性能損失了很多，主要原因這個協(xié)程返回結(jié)果需要等執(zhí)行完畢任務(wù)。
好比打魚，前一種方式是撒網(wǎng)，然后就完事，不聞不問，時間當(dāng)然快，后一種方式則撒網(wǎng)之后，還得收網(wǎng)，等待收網(wǎng)也是一段時間。當(dāng)然，相比同步的方式還是快了千百倍，畢竟撒網(wǎng)還是比一只只釣比較快。
具體使用何種方式，更多的依賴業(yè)務(wù)，不需要返回值的往往需要處理callback，回調(diào)太多容易暈菜，當(dāng)然如果需要很多回調(diào)嵌套，首先優(yōu)化的應(yīng)該是業(yè)務(wù)或產(chǎn)品邏輯。yield的方式很優(yōu)雅，寫法可以異步邏輯同步寫，爽是爽了，當(dāng)然也會損失一定的性能。
異步多樣化
Tornado異步服務(wù)的處理大抵如此。現(xiàn)在異步處理的框架和庫也很多，借助redis或者celery等，也可以把tonrado中一些業(yè)務(wù)異步化，放到后臺執(zhí)行。
此外，Tornado還有客戶端異步功能。該特性主要是在于 AsyncHTTPClient的使用。此時的應(yīng)用場景往往是tornado服務(wù)內(nèi)，需要針對另外的IO進(jìn)行請求和處理。順便提及，上述的例子中，調(diào)用ping其實也算是一種服務(wù)內(nèi)的IO處理。接下來，將會探索一下AsyncHTTPClient的使用，尤其是使用AsyncHTTPClient上傳文件與轉(zhuǎn)發(fā)請求。
異步客戶端
前面了解Tornado的異步任務(wù)的常用做法，姑且歸結(jié)為異步服務(wù)。通常在我們的服務(wù)內(nèi)，還需要異步的請求第三方服務(wù)。針對HTTP請求，Python的庫Requests是最好用的庫，沒有之一。官網(wǎng)宣稱：HTTP for Human。然而，在tornado中直接使用requests將會是一場惡夢。requests的請求會block整個服務(wù)進(jìn)程。
上帝關(guān)上門的時候，往往回打開一扇窗。Tornado提供了一個基于框架本身的異步HTTP客戶端（當(dāng)然也有同步的客戶端）--- AsyncHTTPClient。
AsyncHTTPClient 基本用法
AsyncHTTPClient是 tornado.httpclinet 提供的一個異步http客戶端。使用也比較簡單。與服務(wù)進(jìn)程一樣，AsyncHTTPClient也可以callback和yield兩種使用方式。前者不會返回結(jié)果，后者則會返回response。
如果請求第三方服務(wù)是同步方式，同樣會殺死性能。
|
1
2
3
4
5
6
7
8
|
class SyncHandler(tornado.web.RequestHandler): def get(self, *args, **kwargs): url = 'https://api.github.com/' resp = requests.get(url) print resp.status_code self.finish('It works') |
使用ab測試大概如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
Document Path: /syncDocument Length: 5 bytesConcurrency Level: 5Time taken for tests: 10.255 secondsComplete requests: 5Failed requests: 0Total transferred: 985 bytesHTML transferred: 25 bytesRequests per second: 0.49 [#/sec] (mean)Time per request: 10255.051 [ms] (mean)Time per request: 2051.010 [ms] (mean, across all concurrent requests)Transfer rate: 0.09 [Kbytes/sec] received |
性能相當(dāng)慢了，換成AsyncHTTPClient再測：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
class AsyncHandler(tornado.web.RequestHandler): @tornado.web.asynchronous def get(self, *args, **kwargs): url = 'https://api.github.com/' http_client = tornado.httpclient.AsyncHTTPClient() http_client.fetch(url, self.on_response) self.finish('It works') @tornado.gen.coroutine def on_response(self, response): print response.code |
qps 提高了很多
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
Document Path: /asyncDocument Length: 5 bytesConcurrency Level: 5Time taken for tests: 0.162 secondsComplete requests: 5Failed requests: 0Total transferred: 985 bytesHTML transferred: 25 bytesRequests per second: 30.92 [#/sec] (mean)Time per request: 161.714 [ms] (mean)Time per request: 32.343 [ms] (mean, across all concurrent requests)Transfer rate: 5.95 [Kbytes/sec] received |
同樣，為了獲取response的結(jié)果，只需要yield函數(shù)。
|
1
2
3
4
5
6
7
8
9
10
|
class AsyncResponseHandler(tornado.web.RequestHandler): @tornado.web.asynchronous @tornado.gen.coroutine def get(self, *args, **kwargs): url = 'https://api.github.com/' http_client = tornado.httpclient.AsyncHTTPClient() response = yield tornado.gen.Task(http_client.fetch, url) print response.code print response.body |
AsyncHTTPClient 轉(zhuǎn)發(fā)
使用Tornado經(jīng)常需要做一些轉(zhuǎn)發(fā)服務(wù)，需要借助AsyncHTTPClient。既然是轉(zhuǎn)發(fā)，就不可能只有g(shù)et方法，post，put，delete等方法也會有。此時涉及到一些 headers和body，甚至還有https的waring。
下面請看一個post的例子， yield結(jié)果，通常，使用yield的時候，handler是需要 tornado.gen.coroutine。
|
1
2
3
4
5
6
7
8
9
10
11
|
headers = self.request.headersbody = json.dumps({'name': 'rsj217'})http_client = tornado.httpclient.AsyncHTTPClient()resp = yield tornado.gen.Task( self.http_client.fetch, url, method="POST", headers=headers, body=body, validate_cert=False) |
AsyncHTTPClient 構(gòu)造請求
如果業(yè)務(wù)處理并不是在handlers寫的，而是在別的地方，當(dāng)無法直接使用tornado.gen.coroutine的時候，可以構(gòu)造請求，使用callback的方式。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
body = urllib.urlencode(params)req = tornado.httpclient.HTTPRequest( url=url, method='POST', body=body, validate_cert=False) http_client.fetch(req, self.handler_response)def handler_response(self, response): print response.code |
用法也比較簡單，AsyncHTTPClient中的fetch方法，第一個參數(shù)其實是一個HTTPRequest實例對象，因此對于一些和http請求有關(guān)的參數(shù)，例如method和body，可以使用HTTPRequest先構(gòu)造一個請求，再扔給fetch方法。通常在轉(zhuǎn)發(fā)服務(wù)的時候，如果開起了validate_cert，有可能會返回599timeout之類，這是一個warning，官方卻認(rèn)為是合理的。
AsyncHTTPClient 上傳圖片
AsyncHTTPClient 更高級的用法就是上傳圖片。例如服務(wù)有一個功能就是請求第三方服務(wù)的圖片OCR服務(wù)。需要把用戶上傳的圖片，再轉(zhuǎn)發(fā)給第三方服務(wù)。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
|
@router.Route('/api/v2/account/upload')class ApiAccountUploadHandler(helper.BaseHandler): @tornado.gen.coroutine @helper.token_require def post(self, *args, **kwargs): upload_type = self.get_argument('type', None) files_body = self.request.files['file'] new_file = 'upload/new_pic.jpg' new_file_name = 'new_pic.jpg' # 寫入文件 with open(new_file, 'w') as w: w.write(file_['body']) logging.info('user {} upload {}'.format(user_id, new_file_name)) # 異步請求 上傳圖片 with open(new_file, 'rb') as f: files = [('image', new_file_name, f.read())] fields = (('api_key', KEY), ('api_secret', SECRET)) content_type, body = encode_multipart_formdata(fields, files) headers = {"Content-Type": content_type, 'content-length': str(len(body))} request = tornado.httpclient.HTTPRequest(config.OCR_HOST, method="POST", headers=headers, body=body, validate_cert=False) response = yield tornado.httpclient.AsyncHTTPClient().fetch(request)def encode_multipart_formdata(fields, files): """ fields is a sequence of (name, value) elements for regular form fields. files is a sequence of (name, filename, value) elements for data to be uploaded as files. Return (content_type, body) ready for httplib.HTTP instance """ boundary = '----------ThIs_Is_tHe_bouNdaRY_$' crlf = '\r\n' l = [] for (key, value) in fields: l.append('--' + boundary) l.append('Content-Disposition: form-data; name="%s"' % key) l.append('') l.append(value) for (key, filename, value) in files: filename = filename.encode("utf8") l.append('--' + boundary) l.append( 'Content-Disposition: form-data; name="%s"; filename="%s"' % ( key, filename ) ) l.append('Content-Type: %s' % get_content_type(filename)) l.append('') l.append(value) l.append('--' + boundary + '--') l.append('') body = crlf.join(l) content_type = 'multipart/form-data; boundary=%s' % boundary return content_type, bodydef get_content_type(filename): import mimetypes return mimetypes.guess_type(filename)[0] or 'application/octet-stream' |
對比上述的用法，上傳圖片僅僅是多了一個圖片的編碼。將圖片的二進(jìn)制數(shù)據(jù)按照multipart 方式編碼。編碼的同時，還需要把傳遞的相關(guān)的字段處理好。相比之下，使用requests 的方式則非常簡單：
|
1
2
3
4
5
6
7
|
files = {}f = open('/Users/ghost/Desktop/id.jpg')files['image'] = fdata = dict(api_key='KEY', api_secret='SECRET')resp = requests.post(url, data=data, files=files)f.close()print resp.status_Code |
總結(jié)
通過AsyncHTTPClient的使用方式，可以輕松的實現(xiàn)handler對第三方服務(wù)的請求。結(jié)合前面關(guān)于tornado異步的使用方式。無非還是兩個key。是否需要返回結(jié)果，來確定使用callback的方式還是yield的方式。當(dāng)然，如果不同的函數(shù)都yield，yield也可以一直傳遞。這個特性，tornado的中的tornado.auth 里面對oauth的認(rèn)證。
大致就是這樣的用法。
