python - Flask:关闭数据库连接的方式，是否能够处理高流量？

Question

我需要创建 20 个 API 来使用我的应用程序。这些 API 将从我的 MySQL 数据库中获取数据。因为我预计高峰时段的流量会相当高(高峰时段只有2小时)，每秒可能会收到30到50个请求。

我关心的是联系。由于一次只允许 3 个连接，使用我关闭游标的方法，数据库将能够处理高流量，并且用户能够毫无问题地访问 20 个 API？我获取的数据很小，因为我的整个文本数据库只有 2MB。

如果这种方法可以处理高流量，我就不会想切换到 sqlalchemy。

# A very simple Flask Hello World app for you to get started with...

from flask import Flask,jsonify,abort,make_response,request,render_template
import MySQLdb
import MySQLdb.cursors

app = Flask(__name__)


@app.route('/MUSIC', methods=['GET'])
def Music():
    db = MySQLdb.connect(host='doreme.mysql.pythonanywhere-services.com',user='doreme',passwd='pw',db='doreme$default',cursorclass=MySQLdb.cursors.DictCursor)
    curs = db.cursor()
    try:
        curs.execute("SELECT * FROM MUSIC")
        a = curs.fetchall()
    except Exception:
        return 'Error: unable to fetch items'
    finally:
        curs.close()
        db.close()
    return jsonify({'Music': a})

@app.route('/MUSICKorea', methods=['GET'])
def MusicKorea():
    db = MySQLdb.connect(host='doreme.mysql.pythonanywhere-services.com',user='doreme',passwd='pw',db='doreme$default',cursorclass=MySQLdb.cursors.DictCursor)
    curs = db.cursor()
    try:
        curs.execute("SELECT * FROM MusicKorea")
        b = curs.fetchall()
    except Exception:
        return 'Error: unable to fetch items'
    finally:
        curs.close()
        db.close()
    #return "hihi"
    return jsonify({'Song': b})

@app.route('/MUSICKorea/<Item>', methods=['GET'])
def Musicitem(Korea):
    db = MySQLdb.connect(host='doreme.mysql.pythonanywhere-services.com',user='doreme',passwd='pw',db='doreme$default',cursorclass=MySQLdb.cursors.DictCursor)
    try:
        curs.execute("SELECT * FROM MUSIC WHERE Song LIKE %s",(Korea,))
        c = curs.fetchall()
    except Exception:
        return 'Error: unable to fetch items'
    finally:
        curs.close()
        db.close()
    return jsonify({'Korea': c})

Answer 1

您可以使用 http 基准测试工具(ab、wrk...列出更多工具 here)轻松测试您的 wsgi 应用程序。

测量 python 函数和/或 mysql 查询所花费的时间(非常简单，timeit 可能更好):

import time
...

@app.route('/MUSICKorea/<Item>', methods=['GET'])
def Musicitem(Korea):
    t = time.time()
    db = MySQLdb.connect(host='doreme.mysql.pythonanywhere-services.com',user='doreme',passwd='pw',db='doreme$default',cursorclass=MySQLdb.cursors.DictCursor)
    print('connecting took %.6f seconds' % (time.time()-t))
    try:
        curs.execute("SELECT * FROM MUSIC WHERE Song LIKE %s",(Korea,))
        c = curs.fetchall()
    except Exception:
        return 'Error: unable to fetch items'
    finally:
        curs.close()
        db.close()
    print('Musicitem took %.6f seconds' % (time.time()-t))
    return jsonify({'Korea': c})