-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathtest.py
85 lines (70 loc) · 2.12 KB
/
test.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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import time
import asyncio
import tornado.gen
import tornado.web
import tornado.ioloop
import tornado.httpserver
from tornado.options import define, options
from tornado.httpclient import HTTPClient, AsyncHTTPClient
from requests import get
settings = {'debug': True}
url = "http://127.0.0.1:8888/"
def synchronous_fetch(url):
print("sync")
try:
http_client = HTTPClient()
time.sleep(5)
response = http_client.fetch(url)
print(response.body)
except Exception as e:
print("Error:" + str(e))
return str(e)
http_client.close()
return response.body
def synchronous_get(url):
response = get(url)
time.sleep(5)
print("sync")
return response.text
async def sleep():
print("start")
await asyncio.sleep(5)
print("end")
async def async_fetch(url):
http_client = AsyncHTTPClient()
response = await http_client.fetch(url)
print("async")
return response.body
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("hello: %s" % self.request.request_time())
self.finish()
print("not finish")
return
class synchronous_fetcher(tornado.web.RequestHandler):
def get(self):
self.write("%s, %s" % (synchronous_fetch(url), self.request.request_time()))
class synchronous_geter(tornado.web.RequestHandler):
def get(self):
self.write("%s, %s" % (synchronous_get(url), self.request.request_time()))
class asynchronous_fetcher_1(tornado.web.RequestHandler):
async def get(self):
body = await async_fetch(url)
for i in range(3):
print("skip %s" % i)
await tornado.gen.sleep(5)
time.sleep(5)
print("end request")
self.write("%s, %s" % (body, self.request.request_time()))
def make_app():
return tornado.web.Application([
(r"/", MainHandler),
(r"/1", asynchronous_fetcher_1),
(r"/2", synchronous_fetcher),
(r"/3", synchronous_geter),
], **settings)
if __name__ == '__main__':
print("begin")
app = make_app()
app.listen(8888)
tornado.ioloop.IOLoop.current().start()