欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

基于tornado---异步并发接口

程序员文章站 2022-03-11 08:36:28
1、目的 由于有多个程序和脚本需要对mysql进行读写数据库,每次在脚本中进行数据库的连接、用cursor进行操作过于麻烦,因此希望可以有一个脚本开放接口,只需要传入sql语句,就可以返回结果回来。因此有需要一个可以支持并发量较大的脚本来进行数据库操作。以上就要求我的接口具有异步非阻塞、在结果返回前 ......

1、目的

  由于有多个程序和脚本需要对mysql进行读写数据库,每次在脚本中进行数据库的连接、用cursor进行操作过于麻烦,因此希望可以有一个脚本开放接口,只需要传入sql语句,就可以返回结果回来。因此有需要一个可以支持并发量较大的脚本来进行数据库操作。以上就要求我的接口具有异步非阻塞、在结果返回前保持长连接、并发大。因此单纯的多线程和协程没办法满足要求,就需要用到tornado框架。

2、程序

    1)客户端代码(通过requests 调用接口)

import reuqests
post = requests.post # post请求方式
def db_query(sql, method='query'):
    db_api = 'http://127.0.0.1:8000/db/api'
    db_base = "dbbase"
    db_ret = post(db_api, data=json_encode({'method': method, 'sql': sql, 'dbbase': db_base, 'pwd': 'password123'}))
    if db_ret.status_code == 200:
        if json_decode(db_ret.text)['status'] == 'true':
            db_ret_data = json_decode(db_ret.text)['data']
            return {'status': 'ok', 'data': db_ret_data, 'err': ''}
        else:
            return {'status': 'false', 'data': [], 'err': json_decode(db_ret.text)['data']}
    else:
        return {'status': 'false', 'data': [], 'err': 'connect error'}

 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 = {
        'dbbase': ('authentic', 'password'), # 数据库user、password
    }


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 = 1234

        self.db = mysqldb.connect(db_host, user, password, dbbase, port=db_port, cursorclass=mysqldb.cursors.dictcursor)

        self.cursor = self.db.cursor()

    def query(self, sql):
        self.cursor.execute(sql)
        return self.cursor.fetchall()

    def commit(self, sql):
        try:
            self.cursor.execute(sql)
            self.db.commit()
            return {'status': true, 'data': ''}
        except exception as e:
            self.db.rollback()
            return {'status': false, 'data': e}

    def close(self):
        self.db.close()


class servicehandler(tornado.web.requesthandler):
    executor = threadpoolexecutor(900)  # 必须定义一个executor的属性,然后run_on_executor装饰器才会游泳。

    @run_on_executor  # 线程内运行;query函数被run_on_executor包裹(语法糖),将该函数的执行传递给线程池executor的线程执行,优化了处理耗时性任务,以致达到不阻塞主线程的效果。
    def query(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  # 异步、协程处理;增加并发量
    def post(self):
        data = tornado.escape.json_decode(self.request.body)  # 获取参数,json.loads()解码
        r = {'status': '', 'data': ''}
        if not 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 = {'status': 'false', 'data': 'password error'}
            elif data['method'] == 'query':

                d = yield self.query(data['dbbase'], 'query', data['sql'])
                r = {'status': 'true', 'data': d}
            elif data['method'] == 'commit':

                db_r = yield self.query(data['dbbase'], 'commit', data['sql'])
                if db_r['status']:
                    r = {'status': 'true', 'data': 'commit sucessful'}
                else:
                    r = {'status': 'false', 'data': db_r['data']}
            else:
                r = {'status': 'false', 'data': 'method invaild'}
        self.write(tornado.escape.json_encode(r))  # 写入返回信息写入response
        self.finish()  # 结束服务

    def get(self):
        return self.post()


if __name__ == "__main__":
    application = tornado.web.application([
        (r"/db/api", servicehandler),  # 路由映射
    ])
    application.listen(8000)  # 监听端口
    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语句和操作方式,返回结果或错误内容