如何为烧瓶应用程序加速 JSON？

Question

我目前正在烧瓶中实现一个 webapp。这是一个对收集到的数据进行可视化的应用程序。每个页面或部分都会有一个 GET 调用，每个调用都会返回一个 JSON 响应，然后将其处理为显示的数据。

当前的问题是在函数返回 JSON 响应之前需要进行一些计算。这会导致某些响应的到达速度比其他响应慢，从而使页面加载速度有点慢。我该如何正确处理这个问题？我已经阅读了烧瓶中的缓存，并想知道这是否是应用程序现在需要的。我还对实现 Redis-Queue 进行了一些研究。我不确定哪种方法是正确的。

任何帮助或见解将不胜感激。提前致谢

Answer 1

这里有一些想法：

如果您用于计算的源数据不太可能经常更改，那么您可以运行一次计算并保存结果。然后，只要源数据保持不变，您就可以直接提供结果。

您可以将结果保存回您的数据库，或者按照您的建议，您可以将它们保存在更快的存储设备中，例如 Redis。根据您的描述，我怀疑不经常进行计算会带来很大的性能提升，相比之下，存储在常规数据库与 Redis 或类似数据库中的差异可能并不显着。

如果数据经常变化，那么您仍然需要经常进行计算。对于这种情况，您可以选择将计算推送给客户端。您的 Flask 应用只需返回 JSON 格式的源数据，然后浏览器就可以在用户的​​计算机上进行处理。

我希望这会有所帮助。

Answer 2

你可以使用 copy_current_request_context 和 Redis、线程

当您需要很长时间才能做出 JSON 响应时，它会很有帮助。 第一个请求可能很慢，但下一个请求会更快。

例子

from datetime import timedelta, datetime
from threading import Thread
from . import dbb, redis_client
from flask import Blueprint, request, jsonify, flash, after_this_request, copy_current_request_context, \
    current_app, send_from_directory
from .models import Shop, Customers
def save_customer_json_to_redis(request):
    response_json = {
        "have_customer": False,
        "status": False,
        "anythingelse": None,
        "message":"False, you have to check..."
    }
    #print(request.data)
    headers = request.headers
    Authorization = headers['Authorization']
    token = Authorization.replace("Bearer", "")
    phone = request.args.get('phone')
    if phone is not None and phone != "":
        print('token', token, "phone", phone)
        now = datetime.utcnow() + timedelta(hours=7)
        shop = Shop.query.filter(Shop.private_token == token, Shop.ended_date > now, Shop.shop_active == True).first()      
        customer = Customers.query.filter_by(shop_id=shop.id, phone=phone).first()
        if customer:
            redis_name = f'{shop.id}_api_v2_customer_phone_{phone}_customer_id_{customer.id}'
            print(redis_name)
            response_json["anythingelse"] = ...# do want you want, it need long time to do
            response_json["status"] = True
            response_json["message"] = "Successful"
            redis_client.set(redis_name, json.dumps(response_json)) #Save JSON to Redis
  
@app.route('/api/v2/customer', methods=['GET'])
def api_customer():
    @copy_current_request_context
    def do_update_customer_to_redis():# this function to save JSON you want to response next time to Redis
        save_customer_json_to_redis(request)
        
      
    Thread(target=do_update_customer_to_redis).start() 
    response_json = {
                  "have_customer": False,
                  "status": False,
                  "anythingelse": {},
                  "message": "False, you have to check..."
                  }
    
    #print(request.data)
    headers = request.headers
    Authorization = headers['Authorization']
    token = Authorization.replace("Bearer", "")
    phone = request.args.get('phone')
    if phone is not None and phone != "":
        print('token', token, "phone", phone)
        now = datetime.utcnow() + timedelta(hours=7)
        shop = Shop.query.filter(Shop.private_token == token, Shop.ended_date > now,Shop.shop_active == True).first()
        customer = Customers.query.filter_by(shop_id=shop.id, phone=phone).first()
        if customer:
            redis_name = f'{shop.id}_api_v2_customer_phone_{phone}_customer_id_{customer.id}'
            print(redis_name)
            try:
                response_json = json.loads(redis_client.get(redis_name)) # if have json from app
                print("json.loads(redis_client.get(redis_name))")
            except Exception as e:
                print("json.loads(redis_client.get(redis_name))", e)
                #do any thing you want to response json
                response_json["anythingelse"] = ...# do want you want, it need long time to do
                response_json["message"]= ...#do want you want
                #redis_client.set(redis_name, json.dumps(response_json))
                response_json["status"] = True
                response_json["message"] = "Successful"
   return jsonify(response_json)
                

在init.py

中

from flask import Flask
from flask_cors import CORS
from flask_mail import Mail
from flask_sqlalchemy import SQLAlchemy
from redis import Redis

# init SQLAlchemy so we can use it later in our models
dbb = SQLAlchemy(session_options={"autoflush": False})

redis_client = Redis(
    host='localhost',
    port='6379',
    password='your_redis_password'
)
def create_app():
    app = Flask(__name__)
    ...........