forked from rfyiamcool/tornado-threadpool
-
Notifications
You must be signed in to change notification settings - Fork 0
/
example.py
66 lines (54 loc) · 1.72 KB
/
example.py
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
#coding:utf-8
import time
import logging
import tornado.httpserver
import tornado.ioloop
import tornado.web
from thread_pool import in_thread_pool, in_ioloop, blocking
from tornado.options import define, options
define("port", default=8002, help="run on the given port", type=int)
class TestThreadPool(tornado.web.RequestHandler):
def get(self):
self.blocking_method(self.sleep)
logging.info('end')
self.write('go')
@in_thread_pool
def blocking_method(self,callback):
logging.info('pass thread_pool')
callback()
def sleep(self):
time.sleep(5)
logging.info('this func is sleep')
class TestIoloop(tornado.web.RequestHandler):
def get(self):
self.blocking_method(self.sleep)
logging.info('end')
self.write('go')
@in_ioloop
def blocking_method(self,callback):
logging.info('pass in_ioloop')
callback()
def sleep(self):
time.sleep(5)
logging.info('this func is sleep')
class TestBlocking(tornado.web.RequestHandler):
def get(self):
self.blocking_method(self.sleep)
logging.info('end')
self.write('TestBlocking')
@blocking
def blocking_method(self,callback):
logging.info('pass blocking')
callback()
def sleep(self):
time.sleep(5)
logging.info('this func is sleep')
if __name__ == "__main__":
tornado.options.parse_command_line()
app = tornado.web.Application(handlers=[
(r"/threadpool", TestThreadPool),
(r"/ioloop",TestIoloop ),
(r"/blocking", TestBlocking), ])
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
tornado.ioloop.IOLoop.instance().start()