当前位置：首页 > news >正文

Python Flask内存泄漏分析定位

news 2024/10/28 15:11:43

通过蓝图注册内存跟踪分析接口

示例代码如下，开启内存分析跟踪需要在Flask服务启动前注入环境变量
export PYTHONTRACEMALLOC=1

from typing import Literal
from flask import Blueprint, jsonify, request
import tracemalloc
import os
import linecachesnapshot = None
run_path = os.path.dirname(__file__) # 根据实际需要修改为项目目录，目的是方便后续只追踪本项目的内存泄漏app_bus_blueprint = Blueprint('memory', __name__)def filter_traces(snapshot: tracemalloc.Snapshot, left_trace:Literal['only my code', 'all', 'beside my code']='only my code'):filter_list = (# tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),# tracemalloc.Filter(False, "<unknown>"),# tracemalloc.Filter(False, "<frozen importlib._bootstrap_external>"),tracemalloc.Filter(False, tracemalloc.__file__),tracemalloc.Filter(False, linecache.__file__),tracemalloc.Filter(False, f"*.vscode-server*"),# tracemalloc.Filter(True, jober.__file__),)trace_store = {'only my code': (tracemalloc.Filter(True, f"{run_path}*"),),'all': (),'beside my code':(tracemalloc.Filter(False, f"{run_path}*"),)}filter_list = filter_list + trace_store[left_trace]snapshot = snapshot.filter_traces(filter_list)return snapshotdef display_top(snapshot: tracemalloc.Snapshot, key_type='lineno', limit=30):lines = []# snapshot = filter_traces(snapshot)top_stats = snapshot.statistics(key_type)lines.append("Top %s lines" % limit)for index, stat in enumerate(top_stats[:limit], 1):frame = stat.traceback[0]lines.append("#%s: %s:%s: %.1f KiB"% (index, frame.filename, frame.lineno, stat.size / 1024))# lines.extend(stat.traceback.format())line = linecache.getline(frame.filename, frame.lineno).strip()if line:lines.append('    %s' % line)lines.append('---------------------------------')other = top_stats[limit:]if other:size = sum(stat.size for stat in other)lines.append("%s other: %.1f KiB" % (len(other), size / 1024))total = sum(stat.size for stat in top_stats)lines.append("Total allocated size: %.1f KiB" % (total / 1024))return linesif str(os.getenv('PYTHONTRACEMALLOC', 0)) == '1':tracemalloc.start(25)@app_bus_blueprint.route('/memory_snapshot', methods=['POST', 'GET'], strict_slashes=False)def memory_snapshot():is_compare = isinstance(request.args.get('compare', False), str)global snapshotif not snapshot:snapshot = tracemalloc.take_snapshot()snapshot = filter_traces(snapshot)return "Taken snapshot."else:lines = []snapshot_current = tracemalloc.take_snapshot()snapshot_current = filter_traces(snapshot_current)if is_compare:top_stats = snapshot_current.compare_to(snapshot, 'lineno')# 过滤出只有增长的内存分配increased_stats = [stat for stat in top_stats if stat.size_diff > 0]# 取出增长最多的前10条数据top_increased_stats = sorted(increased_stats, key=lambda stat: stat.size_diff, reverse=True)for stat in top_increased_stats[:20]:lines.append("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))lines.extend(stat.traceback.format())# lines.append(str(stat))lines.append('-----------------------------------')total_increased = sum(stat.size_diff for stat in top_increased_stats)total_decreased = sum(stat.size_diff for stat in top_stats if stat.size_diff < 0)totol_allocated = sum(stat.size for stat in top_stats)lines.append(f"Total increased size: {total_increased / 1024:.1f} KiB")lines.append(f"Total decreased size: {total_decreased / 1024:.1f} KiB")lines.append(f"Absolute change size: {(total_increased + total_decreased) / 1024:.1f} KiB")lines.append(f"Total allocated size: {totol_allocated / 1024:.1f} KiB")else:lines = display_top(snapshot_current, key_type='traceback')snapshot = snapshot_currentreturn jsonify(lines)