anic是一個支持 async/await 語法的異步無阻塞框架，Flask的升級版，效率更高，性能會提升不少，我將同一服務分別用Flask和Sanic編寫，再將壓測的結果進行對比，發現Sanic編寫的服務大概是Falsk的1.5倍。
不過Sanic對環境的要求比較苛刻：linux /Mac + python3.5+
window不支持uvloop
先上一個簡單案例：
#!/usr/bin/env python from sanic import Sanic from sanic.response import text app = Sanic() @app.route("/") async def test(request): return text('Hello World!') if __name__ == "__main__": app.run(host="0.0.0.0", port=8000)
來對比一下flask的code:
from flask import Flask from flask.ext import restful app = Flask(__name__) api = restful.Api(app) class HelloWorld(restful.Resource): def get(self): return {'hello': 'world'} api.add_resource(HelloWorld, '/') if __name__ == '__main__': app.run(debug=True)
從兩者對比，可以看到相似性非常高，可以作為flask是完全沒有問題的。
一、Snaic基本功能
這里筆者只解讀了Snaic的三個方面：Request 、Routing、Response。
1.Routing路由
一個簡單的例子：
@app.route('/post7/<param>', methods=['POST','GET'], host='example.com')- 1
‘/post7’代表接下來post時候，需要在url后面加上這個后綴：’http://127.0.0.1:8000/post7’
methods是指request的方式接受那些方式，常見的有post/ get（大寫）
1.1 傳入參數且參數格式規定
from sanic.response import text @app.route('/tag/<tag>') async def tag_handler(request, tag): return text('Tag - {}'.format(tag))
這邊在URL中會寫入一個參數，’http://127.0.0.1:8000/tag/tag01‘，async def tag_handler之中需要寫入tag參數
然后該參數即可在函數中任意使用。
相似寫法：
@app.route('/number/<integer_arg:int>') async def integer_handler(request, integer_arg): return text('Integer - {}'.format(integer_arg)) @app.route('/number/<number_arg:number>') async def number_handler(request, number_arg): return text('Number - {}'.format(number_arg)) @app.route('/person/<name:[A-z]+>') async def person_handler(request, name): return text('Person - {}'.format(name)) @app.route('/folder/<folder_id:[A-z0-9]{0,4}>') async def folder_handler(request, folder_id): return text('Folder - {}'.format(folder_id))
1.2 路由的第二種寫法
之前看到的路由的寫法都是以裝飾器的形式出現：
@app.route() async def function():
其實也可以不用裝飾器，簡單的寫法：
from sanic.response import text # Define the handler functions async def person_handler2(request, name): return text('Person - {}'.format(name)) # Add each handler function as a route app.add_route(person_handler2, '/person/<name:[A-z]>', methods=['GET'])
通過app.add_route的方式去加路由。
2.Request 請求
來看一個比較完整的例子。
# 加載 import aiohttp import random from sanic import Sanic from sanic.exceptions import NotFound from sanic.response import html, json, redirect app = Sanic() #定義 @app.route('/matt01') async def index_json(request): # 用戶定義一些傳入參數 content = request.args.get('titles') content _list = request.args.getlist('titles') # 獲取數據 return json({'titles':content ,'title_list':content _list,'args1':request.args['titles'], "args2": request.args, "url": request.url, "query_string": request.query_string }) # 啟動 if __name__ == "__main__": app.run(host="0.0.0.0", port=8000)
request.args.get—->得到{‘titles’: [‘yes! hello’,’no!’]}中的第一個’yes! hello’，
request.args.getlist—->得到list所有內容[‘yes! hello’, ‘no!’]，
request.args[‘titles’],—->得到[‘yes! hello’,’no!’]，
request.args—->得到{‘titles’: [‘yes! hello’, ‘no!’]}，
request.url—->傳入URL的所有內容，
request.query_string—->IP+端口+Routing之后的所有內容，
有兩種獲取結果的寫法：
‘http://127.0.0.1:8000/matt01‘后面可以直接接dict，也可以用?后面接上。
get('http://127.0.0.1:8000/matt01',{'titles': ['yes! hello','no!']}).json() >>> {'args1': ['yes! hello', 'no!'], >>> 'args2': {'titles': ['yes! hello', 'no!']}, >>> 'query_string': 'titles=yes%21+hello&titles=no%21', >>> 'title_list': ['yes! hello', 'no!'], >>> 'titles': 'yes! hello', >>> 'url': 'http://127.0.0.1:8000/matt01?titles=yes%21+hello&titles=no%21'} get('http://127.0.0.1:8000/matt01?titles=value1&titles=value2').json() >>> {'args1': ['value1', 'value2'], >>> 'args2': {'titles': ['value1', 'value2']}, >>> 'query_string': 'titles=value1&titles=value2', >>> 'title_list': ['value1', 'value2'], >>> 'titles': 'value1', >>> 'url': 'http://127.0.0.1:8000/matt01?titles=value1&titles=value2'}3.Response
在Request 之中，較多的都是以json格式，也可以是很多其他的格式：text、HTML、file、Streaming等。
3.1 文本格式
from sanic import response @app.route('/text') def handle_request(request): return response.text('Hello world!')
3.2 HTML
from sanic import response @app.route('/html') def handle_request(request): return response.html('<p>Hello world!</p>')
3.3 JSON
from sanic import response @app.route('/json') def handle_request(request): return response.json({'message': 'Hello world!'})
3.4 File
from sanic import response @app.route('/file') async def handle_request(request): return await response.file('/srv/www/whatever.png')
案例一：回傳圖片案例
回傳圖片用的是response.file
# 實驗8：傳回一張圖 from sanic import response @app.route('/file8') async def handle_request(request): return await response.file('/mnt/flask/out3.jpg')
# 實驗8 返回圖片 get('http://127.0.0.1:8000/file8') >>> <Response [200]>
返回的是bytes格式的圖片。
.
二、Snaic其他信息
1.app.run參數
來源於：Sanic框架
try: serve( host=host, port=port, debug=debug, # 服務開始后啟動的函數 after_start=after_start, # 在服務關閉前啟動的函數 before_stop=before_stop, # Sanic(__name__).handle_request() request_handler=self.handle_request, # 默認讀取Config request_timeout=self.config.REQUEST_TIMEOUT, request_max_size=self.config.REQUEST_MAX_SIZE, ) except: pass
- host(默認“127.0.0.1”): 服務器主機的地址。
- port(默認8000): 服務器的端口。
- debug(默認False): 啟用調試（減慢服務器速度）。
- ssl(默認None): 用於工作者SSL加密的SSLContext。
- sock（默認None）：服務器接受連接的Socket。
- worker（默認值1）：生成的工作進程數。
- loop（默認None）: asyncio兼容的事件循環。如果沒有指定，Sanic會創建自己的事件循環。
- protocol（默認HttpProtocol）：asyncio.protocol的子類。
默認情況下，Sanic在主進程中只偵聽一個CPU內核。要啟動其它核心，只需指定run參數中進程的數量。
app.run(host='0.0.0.0', port=1337, workers=4)
Sanic將自動啟動多個進程並在它們之間建立路由路徑。建議進程數和CPU核心數一樣。
命令行格式運行：
如果你將Sanic應用程序在名為server.py的文件中初始化，那么可以像這樣運行服務：
python -m sanic server.app --host=0.0.0.0 --port=1337 --workers=4
2.報錯信息的返回
from sanic.response import text from sanic.exceptions import NotFound @app.exception(NotFound) def ignore_404s(request, exception): return text("Yep, I totally found the page: {}".format(request.url))
如果出現錯誤就會顯示出以下的內容：
b'Yep, I totally found the page: http://127.0.0.1:8000/users/matt'
3.藍本 Blueprint
把一些小功能包裝在一個小集合Blueprint里面。參考
from sanic import Blueprint from sanic.response import html, json, redirect app = Sanic() blueprint = Blueprint('name', url_prefix='/my_blueprint') blueprint2 = Blueprint('name2', url_prefix='/my_blueprint2') @blueprint.route('/foo') async def foo(request): return json({'msg': 'hi from blueprint'}) @blueprint2.route('/foo') async def foo2(request): return json({'msg': 'hi from blueprint2'}) app.register_blueprint(blueprint) app.register_blueprint(blueprint2) app.run(host="0.0.0.0", port=8000, debug=True)
Blueprint(‘name’, url_prefix=’/my_blueprint’)中為，該小藍本的名字為‘name’，前綴為：’/my_blueprint’;
定義藍本之后，就要定義藍本內容，@blueprint.route('/foo')還有@blueprint2.route('/foo')，即為相關內容；
app.register_blueprint(blueprint)和app.register_blueprint(blueprint2)把藍本合並到app這個大服務中，然后一起打包給出。
因為每個藍本有不同的前綴，所以需要在URL之后加入自己的前綴內容以示區分：
get('http://0.0.0.0:8000/my_blueprint2/foo').content >>> b'{"msg":"hi from blueprint2"}' get('http://0.0.0.0:8000/my_blueprint/foo').content >>> b'{"msg":"hi from blueprint"}'
延伸一：路由為post，如何寫請求？
對於小白來說，post的方式太難懂了。
@app.route('/post12', methods=['POST']) async def get_handler(request): return json('POST request - {}'.format(request.body)) # 啟動 if __name__ == "__main__": app.run(host="0.0.0.0", port=8000)
這邊用request.get 或者request.args好像在post都不太好用，只能在get請求上可以使用。這邊只有request.body比較使用。
那么Post以及Post之后的結果可見：
post('http://127.0.0.1:8000/post12',{'titles': ['value1', 'value2']}).json() >>> "POST request - b'titles=value1&titles=value2'" post('http://127.0.0.1:8000/post12','value1').json() >>> "POST request - b'value1'" # 實踐不出來的幾種方式 post('http://127.0.0.1:8000/post12?titles=value1&titles=value2').json() >>> 'POST request - None'
延伸二：設置sanic 的HTTPS服務
網上教程很多，利用 stunnel 建立一個 SSL 加密通道綁定到 Django/Flask 內建服務器，好像不好用。
HTTPS相當於需要設置SSL證書，sanic官方文檔寫的是：
ssl = {'cert': "/path/to/cert", 'key': "/path/to/keyfile"} app.run(host="0.0.0.0", port=8443, ssl=ssl)
那么這里可以使用openssl，就需要設置一步：
openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365
然后依次輸入：
> openssl req -new -out ca/ca-req.csr -key ca/ca-key.pem
Country Name (2 letter code) [AU]:cn State or Province Name (full name) [Some-State]:guangdong Locality Name (eg, city) []:guangzhou Organization Name (eg, company) [Internet Widgits Pty Ltd]:test Organizational Unit Name (eg, section) []:test Common Name (eg, YOUR name) []:root Email Address []:test
一些個人信息都可以隨意填時該目就會生成： key.pem,key.pem
ssl = {'cert': "./cert.pem", 'key': "./key.pem"} app.run(host="0.0.0.0", port=xxxx,ssl = ssl)
延伸三：壓力測試
測試環節有兩個部分：單元測試與壓力測試。單元測試這邊其實可簡單可復雜，sanic有自己的測試環節，插件：pytest，這邊提一下壓力測試。使用的是：locust，壓力測試最好在內外網都進行測試下，當然服務器配置是你定。（主要參考：9. 測試與部署）
from requests import post,get from locust import TaskSet, task from locust import HttpLocust import sys # 端口調用環節 def action_rss(client): data = '心心念念的要來吃花和，好不容易找到個周日的中午有時間，氣溫剛好不熱，組個小團來吃頓棒棒的海鮮,味道超級好。'.encode("utf-8") # matt route = '/seg' # matt headers = {'content-type': 'application/json'} return client.post(route, data =data,headers=headers).json() # 壓力測試 class RssBehavior(TaskSet): @task(1) def interface_rss(self): action_rss(self.client) # locust class Rss(HttpLocust): host = 'http://0.0.0.0:7777' # matt task_set = RssBehavior min_wait = 20 max_wait = 3000
執行的時候，在終端輸入：locust -f locustfile.py --no-web -c 1，這個結果直接在終端顯示，也可以使用：locust -f locustfile.py，可以通過訪問IP：http://0.0.0.0:8089/，在web端口看到壓力測試的結果。
再來說一下，里面的邏輯，
-
class Rss(HttpLocust):為主函數
min_wait/max_wait為最小、最大重復測試次數，host為主要請求的端口，此時不能加路由route class -
RssBehavior(TaskSet)執行函數
筆者之前沒有接觸這樣的模塊，這段代碼這段很詭異的是，一開始筆者很難理解，self.client在哪定義了？？
不過，原來在這段函數中，只要執行了Rss，那么隱藏了這么一條信息：self.client =
'http://0.0.0.0:7777' action_rss(client)比較奇怪的就是：client.pos環節，那么這邊隱藏的條件為：還有路由'/seg'和數據data其實都可以及時變更。
client.post(route, data =data,headers=headers).json() = 'http://0.0.0.0:7777'.post('/seg',data = data)