Python Flask内存泄漏分析定位

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

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

1. from typing import Literal
2. from flask import Blueprint, jsonify, request
3. import tracemalloc
4. import os
5. import linecache
7. snapshot = None
8. run_path = os.path.dirname(__file__) # 根据实际需要修改为项目目录，目的是方便后续只追踪本项目的内存泄漏
11. app_bus_blueprint = Blueprint('memory', __name__)
13. def filter_traces(snapshot: tracemalloc.Snapshot, left_trace:Literal['only my code', 'all', 'beside my code']='only my code'):
14.     filter_list = (
15.         # tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),
16.         # tracemalloc.Filter(False, "<unknown>"),
17.         # tracemalloc.Filter(False, "<frozen importlib._bootstrap_external>"),
18.         tracemalloc.Filter(False, tracemalloc.__file__),
19.         tracemalloc.Filter(False, linecache.__file__),
20.         tracemalloc.Filter(False, f"*.vscode-server*"),
21.         # tracemalloc.Filter(True, jober.__file__),
22.     )
23.     trace_store = {
24.         'only my code': (
25.             tracemalloc.Filter(True, f"{run_path}*"),
26.         ),
27.         'all': (),
28.         'beside my code':(
29.             tracemalloc.Filter(False, f"{run_path}*"),
30.         )
31.     }
32.     filter_list = filter_list + trace_store[left_trace]
33.     snapshot = snapshot.filter_traces(filter_list)
34.     return snapshot
36. def display_top(snapshot: tracemalloc.Snapshot, key_type='lineno', limit=30):
37.     lines = []
38.     # snapshot = filter_traces(snapshot)
39.     top_stats = snapshot.statistics(key_type)
40.     lines.append("Top %s lines" % limit)
41.     for index, stat in enumerate(top_stats[:limit], 1):
42.         frame = stat.traceback[0]
43.         lines.append("#%s: %s:%s: %.1f KiB"
44.               % (index, frame.filename, frame.lineno, stat.size / 1024))
45.         # lines.extend(stat.traceback.format())
46.         line = linecache.getline(frame.filename, frame.lineno).strip()
47.         if line:
48.             lines.append('    %s' % line)
49.         lines.append('---------------------------------')
50.     other = top_stats[limit:]
51.     if other:
52.         size = sum(stat.size for stat in other)
53.         lines.append("%s other: %.1f KiB" % (len(other), size / 1024))
54.     total = sum(stat.size for stat in top_stats)
55.     lines.append("Total allocated size: %.1f KiB" % (total / 1024))
56.     return lines
58. if str(os.getenv('PYTHONTRACEMALLOC', 0)) == '1':
59.     tracemalloc.start(25)
60.     @app_bus_blueprint.route('/memory_snapshot', methods=['POST', 'GET'], strict_slashes=False)
61.     def memory_snapshot():
62.         is_compare = isinstance(request.args.get('compare', False), str)
63.         global snapshot
64.         if not snapshot:
65.             snapshot = tracemalloc.take_snapshot()
66.             snapshot = filter_traces(snapshot)
67.             return "Taken snapshot."
68.         else:
69.             lines = []
70.             snapshot_current = tracemalloc.take_snapshot()
71.             snapshot_current = filter_traces(snapshot_current)
72.             if is_compare:
73.                 top_stats = snapshot_current.compare_to(snapshot, 'lineno')
74.                 # 过滤出只有增长的内存分配
75.                 increased_stats = [stat for stat in top_stats if stat.size_diff > 0]
76.                 # 取出增长最多的前10条数据
77.                 top_increased_stats = sorted(increased_stats, key=lambda stat: stat.size_diff, reverse=True)
78.                 for stat in top_increased_stats[:20]:
79.                     lines.append("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
80.                     lines.extend(stat.traceback.format())
81.                     # lines.append(str(stat))
82.                     lines.append('-----------------------------------')
83.                 total_increased = sum(stat.size_diff for stat in top_increased_stats)
84.                 total_decreased = sum(stat.size_diff for stat in top_stats if stat.size_diff < 0)
85.                 totol_allocated = sum(stat.size for stat in top_stats)
86.                 lines.append(f"Total increased size: {total_increased / 1024:.1f} KiB")
87.                 lines.append(f"Total decreased size: {total_decreased / 1024:.1f} KiB")
88.                 lines.append(f"Absolute change size: {(total_increased + total_decreased) / 1024:.1f} KiB")
89.                 lines.append(f"Total allocated size: {totol_allocated / 1024:.1f} KiB")
90.             else:
91.                 lines = display_top(snapshot_current, key_type='traceback')
92.             snapshot = snapshot_current
93.             return jsonify(lines)

复制代码

通过trace filter，可以选择’only my code’, ‘all’, 'beside my code’三种trace筛选策略，意思为：只跟踪我的工作区代码，所有，非我的代码/第三方包。
参考文章：

• 获取一个内存块的溯源；
  
• 定位python内存泄漏问题
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。