高频POST请求存储问题：基于Flask接收BLE网关数据的技术问询

一、完善基础逻辑：添加存储与错误处理 #

现有代码仅获取了数据但未执行存储操作，且缺少异常处理机制，先修复基础功能：

from flask import Flask, request, abort
import json
import time

app = Flask(__name__)

# 示例：本地文件存储（生产环境建议替换为数据库）
def save_data(data):
    # 追加写入日志文件，避免单文件过大
    with open("ble_data.log", "a", encoding="utf-8") as f:
        json.dump(data, f)
        f.write("\n")

@app.route('/ble', methods=['POST'])
def ble_packet():
    # 校验请求格式
    if not request.is_json:
        abort(400, description="请求体必须为JSON格式")
    
    try:
        data_packet = request.json.get('data')
        if not data_packet:
            abort(400, description="缺少必填字段'data'")
        
        # 执行存储操作
        save_data(data_packet)
        return {"status": "success"}, 200
    except Exception as e:
        abort(500, description=f"数据处理失败: {str(e)}")

if __name__ == '__main__':
    # 开发模式仅用于调试，生产环境禁用debug
    app.run(port=8080)

二、高频请求性能优化 #

1. 替换Flask开发服务器 #

Flask自带的单线程开发服务器无法承载高频请求，生产环境必须使用高性能WSGI/ASGI服务器：

• Gunicorn：多进程部署，命令示例：

  gunicorn -w 4 -b 0.0.0.0:8080 your_app_file:app
  

  -w指定工作进程数，建议设置为CPU核心数的2倍
• Uvicorn：异步ASGI服务器，配合异步框架（如FastAPI）性能更优

2. 异步处理存储任务 #

同步存储会阻塞请求响应，建议将存储操作放入消息队列异步执行：

• 示例：用Redis+RQ实现任务队列

  from rq import Queue
  from redis import Redis
  
  # 初始化Redis连接与任务队列
  redis_conn = Redis(host='localhost', port=6379)
  q = Queue(connection=redis_conn)
  
  # 定义可被队列调用的存储任务
  def save_data(data):
      with open("ble_data.log", "a", encoding="utf-8") as f:
          json.dump(data, f)
          f.write("\n")
  
  @app.route('/ble', methods=['POST'])
  def ble_packet():
      # ... 前面的校验逻辑 ...
      # 将存储任务放入队列，立即返回响应
      q.enqueue(save_data, data_packet)
      return {"status": "success"}, 200
  

3. 批量存储减少IO开销 #

若业务允许短暂延迟，可缓存多条数据后批量写入，降低磁盘/数据库IO频率：

from collections import deque
import threading
import time

# 内存缓存，最多保存100条数据
data_cache = deque(maxlen=100)
cache_lock = threading.Lock()

def batch_save():
    while True:
        time.sleep(5)  # 每5秒执行一次批量写入
        with cache_lock:
            if len(data_cache) > 0:
                batch_data = list(data_cache)
                data_cache.clear()
                # 批量写入逻辑，示例为文件存储
                with open("ble_batch_data.log", "a", encoding="utf-8") as f:
                    for data in batch_data:
                        json.dump(data, f)
                        f.write("\n")

# 启动批量存储后台线程
threading.Thread(target=batch_save, daemon=True).start()

@app.route('/ble', methods=['POST'])
def ble_packet():
    # ... 校验逻辑 ...
    with cache_lock:
        data_cache.append(data_packet)
    return {"status": "success"}, 200

三、选择适配的存储介质 #

• 时序数据库：如InfluxDB、TimescaleDB，专为时间序列传感器数据设计，写入与查询性能远超普通关系型数据库
• Redis：需实时查询数据时，可先写入Redis缓存，再异步同步至持久化存储
• 文件存储：适合临时存储或小量数据，建议用换行分隔的JSON格式，便于后续解析

内容的提问来源于stack exchange，提问作者Rikissssss