python---tornado補充（異步非阻塞）

時間 2019-11-13

原文原文鏈接

一：正常訪問（同一線程中多個請求是同步阻塞狀態）

import tornado.ioloop
import tornado.web
import tornado.websocket
import datetime,time

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("main")class IndexHandler(tornado.web.RequestHandler):
    def get(self):
        time.sleep(10)
        self.write("index")

st ={
    "template_path": "template",#模板路徑配置
    "static_path":'static',
}

#路由映射   匹配執行，不然404
application = tornado.web.Application([
    ("/main",MainHandler),
    ("/index",IndexHandler),
],**st)

if __name__=="__main__":
    application.listen(8080)

    #io多路複用
    tornado.ioloop.IOLoop.instance().start()

咱們先訪問index，再去訪問main，查看狀況

二：使用future模塊，實現異步非阻塞

import tornado.ioloop
import tornado.web
import tornado.websocket
import time
from tornado.concurrent import Future

class MainHandler(tornado.web.RequestHandler):
    def get(self):

        self.write("main")class IndexHandler(tornado.web.RequestHandler):
    def get(self):
 future = Future() tornado.ioloop.IOLoop.current().add_timeout(time.time()+5,self.done) #會在結束後爲future中result賦值 yield future

    def done(self,*args,**kwargs):
        self.write("index")
        self.finish()　　#關閉請求鏈接，必須在回調中完成

st ={
    "template_path": "template",#模板路徑配置
    "static_path":'static',
}

#路由映射   匹配執行，不然404
application = tornado.web.Application([
    ("/main",MainHandler),
    ("/index",IndexHandler),
],**st)

if __name__=="__main__":
    application.listen(8080)

    #io多路複用
    tornado.ioloop.IOLoop.instance().start()

三：在tornado中使用異步IO請求模塊

import tornado.ioloop
import tornado.web
import tornado.websocket
import time
from tornado.concurrent import Future
from tornado import httpclient from tornado import gen class MainHandler(tornado.web.RequestHandler):
    def get(self):

        self.write("main")

    def post(self, *args, **kwargs):
        pass

class IndexHandler(tornado.web.RequestHandler):
    @gen.coroutine
    def get(self):
        http = httpclient.AsyncHTTPClient()
        yield http.fetch("http://www.google.com",self.done)

    def done(self):
        self.write("index")
        self.finish()

    def post(self, *args, **kwargs):
        pass

st ={
    "template_path": "template",#模板路徑配置
    "static_path":'static',
}

#路由映射   匹配執行，不然404
application = tornado.web.Application([
    ("/main",MainHandler),
    ("/index",IndexHandler),
],**st)

if __name__=="__main__":
    application.listen(8080)

    #io多路複用
    tornado.ioloop.IOLoop.instance().start()

四：請求間交互，使用future

import tornado.ioloop
import tornado.web
import tornado.websocket
from tornado.concurrent import Future
from tornado import gen

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("main")


class IndexHandler(tornado.web.RequestHandler):
 @gen.coroutine
    def get(self):
 future = Future() future.add_done_callback(self.done) yield future #因爲future中的result中值一直未被賦值，全部客戶端一直等待 
    def done(self,*args,**kwargs):
        self.write("index")
        self.finish()

st ={
    "template_path": "template",#模板路徑配置
    "static_path":'static',
}

#路由映射   匹配執行，不然404
application = tornado.web.Application([
    ("/main",MainHandler),
    ("/index",IndexHandler),
],**st)

if __name__=="__main__":
    application.listen(8080)

    #io多路複用
    tornado.ioloop.IOLoop.instance().start()

咱們能夠在另外一個請求中去爲這個future中result賦值，使當前請求返回

import tornado.ioloop
import tornado.web
import tornado.websocket
from tornado.concurrent import Future
from tornado import gen

future = None

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        global future future.set_result(None)　　#爲Future中result賦值
        self.write("main")


class IndexHandler(tornado.web.RequestHandler):
 @gen.coroutine def get(self): global future future = Future() future.add_done_callback(self.done) yield future #因爲future中的result中值一直未被賦值，全部客戶端一直等待

    def done(self,*args,**kwargs):
        self.write("index")
        self.finish()

st ={
    "template_path": "template",#模板路徑配置
    "static_path":'static',
}

#路由映射   匹配執行，不然404
application = tornado.web.Application([
    ("/main",MainHandler),
    ("/index",IndexHandler),
],**st)

if __name__=="__main__":
    application.listen(8080)

    #io多路複用
    tornado.ioloop.IOLoop.instance().start()

五：自定義web框架（同步）

# coding:utf8
# __author:  Administrator
# date:      2018/6/30 0030
# /usr/bin/env python
import socket
from select import select
import re

class HttpResponse(object):
    """
    封裝響應信息
    """
    def __init__(self, content=''):
        self.content = content
        '''

        '''
        self.status = "HTTP/1.1 200 OK"
        self.headers = {}
        self.cookies = {}

        self.initResponseHeader()

    def changeStatus(self,status_code,status_desc):
        self.status = "HTTP/1.1 %s %s"%(status_code,status_desc)

    def initResponseHeader(self):
        self.headers['Content-Type']='text/html; charset=utf-8'
        self.headers['X-Frame-Options']='SAMEORIGIN'
        self.headers['X-UA-Compatible']='IE=10'
        self.headers['Cache-Control']='private, max-age=10'
        self.headers['Vary']='Accept-Encoding'
        self.headers['Connection']='keep-alive'

    def response(self):
        resp_content = None
        header_list = [self.status,]
        for item in self.headers.items():
            header_list.append("%s: %s"%(item[0],item[1]))

        header_str = "\r\n".join(header_list)
        resp_content = "\r\n\r\n".join([header_str,self.content])
        return bytes(resp_content, encoding='utf-8')

class HttpRequest:
    def __init__(self,content):
        """content：用戶傳遞的請求頭信息，字節型"""
        self.content = content
        self.header_bytes = bytes()
        self.body_bytes = bytes()

        self.header_str = ""
        self.body_str = ""

        self.header_dict = {}

        self.method = ""
        self.url = ""
        self.protocol = ""

        self.initialize()
        self.initialize_headers()

    def initialize(self):
        data = self.content.split(b"\r\n\r\n",1)
        if len(data) == 1:  #全是請求頭
            self.header_bytes = self.content
        else:   #含有請求頭和請求體
            self.header_bytes,self.body_bytes = data
        self.header_str = str(self.header_bytes,encoding="utf-8")
        self.body_str = str(self.body_bytes,encoding="utf-8")

    def initialize_headers(self):
        headers = self.header_str.split("\r\n")
        first_line = headers[0].split(" ")
        if len(first_line) == 3:
            self.method,self.url,self.protocol = first_line
        for line in headers[1:]:
            k_v = line.split(":",1)
            if len(k_v) == 2:
                k,v = k_v
                self.header_dict[k] = v


def main(request):
    return "main"

def index(request):
    return "index"

routers = [
    ("/main/",main),
    ('/index/',index),
]

def run():
    sock = socket.socket()
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(('127.0.0.1', 8080))
    sock.listen(128)
    sock.setblocking(False)

    inputs = []
    inputs.append(sock)
    while True:
        rlist, wlist, elist = select(inputs, [], [], 0.05)  # http是單向的，咱們只獲取請求便可
        for r in rlist:
            if r == sock:  # 有新的請求到來
                conn, addr = sock.accept()
                conn.setblocking(False)
                inputs.append(conn)
            else:  # 客戶端請求數據
                data = b""
                # 開始獲取請求頭
                while True:
                    try:
                        chunk = r.recv(1024)
                        data += chunk
                    except BlockingIOError as e:
                        chunk = None
                    if not chunk:
                        break

                # 處理請求頭，請求體
                request = HttpRequest(data)
                #1.獲取url
                #2.路由匹配
                #3.執行函數，獲取返回值
                #4.將返回值發送
                flag = False
                func = None
                for route in routers:
                    if re.match(route[0],request.url):
                        flag = True
                        func = route[1]
                        break
                if flag:
                    result = func(request)
                    response = HttpResponse(result)
                    r.sendall(response.response())
                else:
                    response = HttpResponse("Not Found")
                    response.changeStatus(404,"Not Page")
                    r.sendall(response.response())
                inputs.remove(r)
                r.close()


if __name__ == "__main__":
    run()

未實現異步非阻塞

六：完善自定義web框架（異步）

import socket
from select import select
import re,time

class HttpResponse(object):
    """
    封裝響應信息
    """
    def __init__(self, content=''):
        self.content = content
        '''

        '''
        self.status = "HTTP/1.1 200 OK"
        self.headers = {}
        self.cookies = {}

        self.initResponseHeader()

    def changeStatus(self,status_code,status_desc):
        self.status = "HTTP/1.1 %s %s"%(status_code,status_desc)

    def initResponseHeader(self):
        self.headers['Content-Type']='text/html; charset=utf-8'
        self.headers['X-Frame-Options']='SAMEORIGIN'
        self.headers['X-UA-Compatible']='IE=10'
        self.headers['Cache-Control']='private, max-age=10'
        self.headers['Vary']='Accept-Encoding'
        self.headers['Connection']='keep-alive'

    def response(self):
        resp_content = None
        header_list = [self.status,]
        for item in self.headers.items():
            header_list.append("%s: %s"%(item[0],item[1]))

        header_str = "\r\n".join(header_list)
        resp_content = "\r\n\r\n".join([header_str,self.content])
        return bytes(resp_content, encoding='utf-8')

class HttpRequest:
    def __init__(self,content):
        """content：用戶傳遞的請求頭信息，字節型"""
        self.content = content
        self.header_bytes = bytes()
        self.body_bytes = bytes()

        self.header_str = ""
        self.body_str = ""

        self.header_dict = {}

        self.method = ""
        self.url = ""
        self.protocol = ""

        self.initialize()
        self.initialize_headers()

    def initialize(self):
        data = self.content.split(b"\r\n\r\n",1)
        if len(data) == 1:  #全是請求頭
            self.header_bytes = self.content
        else:   #含有請求頭和請求體
            self.header_bytes,self.body_bytes = data
        self.header_str = str(self.header_bytes,encoding="utf-8")
        self.body_str = str(self.body_bytes,encoding="utf-8")

    def initialize_headers(self):
        headers = self.header_str.split("\r\n")
        first_line = headers[0].split(" ")
        if len(first_line) == 3:
            self.method,self.url,self.protocol = first_line
        for line in headers[1:]:
            k_v = line.split(":",1)
            if len(k_v) == 2:
                k,v = k_v
                self.header_dict[k] = v

class Future:
    def __init__(self,timeout):
        self.result = None
        self.timeout = timeout
        self.start = time.time()

    def add_callback_done(self,callback,request):
        self.callback = callback
        self.request = request

    def call(self):
        if self.result == "timeout":　　#超時就不要去獲取頁面數據，直接返回超時 return "timeout"
        if self.result:　　#如果沒有超時，去獲取回調數據 return self.callback(self.request)

def callback(request):
    print(request)
    return "async main"

f = None

def main(request):
    global f f = Future(10) f.add_callback_done(callback,request)　　#設置回調 return f

def index(request):
    return "index"

def stop(request):
    if f:
        f.result = True
    return "stop"

routers = [
    ("/main/",main),
    ('/index/',index),
    ('/stop/', stop),　　#用於向future的result賦值
]

def run():
    sock = socket.socket()
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(('127.0.0.1', 8080))
    sock.listen(128)
    sock.setblocking(False)

    inputs = []
    async_request_dict = {}
    inputs.append(sock)
    while True:
        rlist, wlist, elist = select(inputs, [], [], 0.05)  # http是單向的，咱們只獲取請求便可
        for r in rlist:
            if r == sock:  # 有新的請求到來
                conn, addr = sock.accept()
                conn.setblocking(False)
                inputs.append(conn)
            else:  # 客戶端請求數據
                data = b""
                # 開始獲取請求頭
                while True:
                    try:
                        chunk = r.recv(1024)
                        data += chunk
                    except BlockingIOError as e:
                        chunk = None
                    if not chunk:
                        break

                # 處理請求頭，請求體
                request = HttpRequest(data)
                #1.獲取url
                #2.路由匹配
                #3.執行函數，獲取返回值
                #4.將返回值發送
                flag = False
                func = None
                for route in routers:
                    if re.match(route[0],request.url):
                        flag = True
                        func = route[1]
                        break
                if flag:
                    result = func(request)
                    if isinstance(result,Future):　　#對於future對象，咱們另外作異步處理，不阻塞當前操做 async_request_dict[r] = result continue
                    response = HttpResponse(result)
                    r.sendall(response.response())
                else:
                    response = HttpResponse("Not Found")
                    response.changeStatus(404,"Not Page")
                    r.sendall(response.response())
                inputs.remove(r)
                r.close()

        for conn in list(async_request_dict.keys()):　　#另外對future對象處理
            future = async_request_dict[conn]
            start = future.start
            timeout = future.timeout
            if (start+timeout) <= time.time():　　#超時檢測
                future.result = "timeout"
            if future.result:
                response = HttpResponse(future.call())　　#獲取回調數據
                conn.sendall(response.response())
                conn.close()
                del async_request_dict[conn]　　#刪除字典中這個連接，和下面inputs列表中連接
                inputs.remove(conn)

if __name__ == "__main__":
    run()

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。