基於tornado 非同步併發介面

1、目的

由於有多個程式和指令碼需要對mysql進行讀寫資料庫，每次在指令碼中進行資料庫的連線、用cursor進行操作過於麻煩，因此希望可以有乙個指令碼開放介面，只需要傳入sql語句，就可以返回結果回來。因此有需要乙個可以支援併發量較大的指令碼來進行資料庫操作。以上就要求我的介面具有非同步非阻塞、在結果返回前保持長連線、併發大。因此單純的多執行緒和協程沒辦法滿足要求，就需要用到tornado框架。

2、程式

1）客戶端**(通過requests 呼叫介面)

import reuqests

post = requests.post # post請求方式

def db_query(sql, method='
query'):
db_api = '
'db_base = "
dbbase
"db_ret = post(db_api, data=json_encode())
if db_ret.status_code == 200:
if json_decode(db_ret.text)['
status
'] == '
true':
db_ret_data = json_decode(db_ret.text)['
data']
return 
else
:            
return 
else
:        
return

2）服務端**(基於tornado框架)

#
coding=utf8
import
mysqldb
import
mysqldb.cursors
import
tornado
import
tornado.ioloop
import
tornado.web
import
tornado.gen
from tornado.concurrent import
run_on_executor
from concurrent.futures import
threadpoolexecutor
class
db_config:
config =
class
db:    
def__init__
(self, dbbase):
user, password =db_config.config.get(dbbase, (none, none))
if user == none or password ==none:
raise exception('
keyerror
', dbbase)
db_host = '
192.168.xx.xx
'db_port = 1234self.db = mysqldb.connect(db_host, user, password, dbbase, port=db_port, cursorclass=mysqldb.cursors.dictcursor)
self.cursor =self.db.cursor()
defquery(self, sql):
self.cursor.execute(sql)
return
self.cursor.fetchall()
defcommit(self, sql):
try:
self.cursor.execute(sql)
self.db.commit()
return 
except
exception as e:
self.db.rollback()
return 
defclose(self):
self.db.close()
class
servicehandler(tornado.web.requesthandler):
executor = threadpoolexecutor(900)  #
必須定義乙個executor的屬性，然後run_on_executor裝飾器才會游泳。
@run_on_executor  
#執行緒內執行；query函式被run_on_executor包裹(語法糖)，將該函式的執行傳遞給執行緒池executor的執行緒執行，優化了處理耗時性任務，以致達到不阻塞主線程的效果。
defquery(self, dbname, method, sql):
db =db(dbname)
ret = ''
if method == '
query':
ret =db.query(sql)
elif method == '
commit':
ret =db.commit(sql)
db.close()
return
ret    @tornado.web.asynchronous  
#保持長連線，直到處理後返回
@tornado.gen.coroutine  #
非同步、協程處理；增加併發量
defpost(self):
data = tornado.escape.json_decode(self.request.body)  #
獲取引數，json.loads()解碼
r = 
ifnot data.get('
pwd'
, none):
r['status
'], r['
data
'] = ('
false
', '
not password')
elif
not data.get('
dbbase
', none):
r['status
'], r['
data
'] = ('
false
', '
not db select')
else
:            
if data['
pwd'] != '
password123)
': #
介面的密碼認證
r = 
elif data['
method
'] == '
query':
d = yield self.query(data['
dbbase
'], '
query
', data['
sql'
])                r = 
elif data['
method
'] == '
commit':
db_r = yield self.query(data['
dbbase
'], '
commit
', data['
sql'
])                
if db_r['
status']:
r = 
else
:                    r = 
else
:                r = 
self.write(tornado.escape.json_encode(r))  
#寫入返回資訊寫入response
self.finish()  #
結束服務
defget(self):
return
self.post()
if__name__ == "
__main__":
(r"/db/api
", servicehandler),  #
路由對映
])監聽埠
tornado.ioloop.ioloop.instance().start()  #
啟動服務

3、請求舉例(post請求)

1) sql語句

sql = '

select vid from video where status=1 group by vid order by num asc limit 100

2)客戶端呼叫db_query函式

db_ret = db_query(sql, '

query

') # 其中query是資料庫操作的方法 query為查詢/commit 為insert/update/delete等資料庫修改操作時使用

通過呼叫db_query傳入引數sql語句和操作方式，返回結果或錯誤內容

基於tornado 非同步併發介面

Tornado非同步學習

Tornado非同步模式

Tornado 非同步socketTCP通訊

基於tornado 非同步併發介面

Tornado非同步學習

Tornado非同步模式

Tornado 非同步socketTCP通訊

相關推薦