告别Print-Python调试入门,用PDB高效找Bug
目录
告别Print: Python调试入门，用PDB高效找Bug
『宝藏代码胶囊开张啦！』—— 我的 CodeCapsule 来咯！✨
写代码不再头疼！我的新站点 CodeCapsule 主打一个 “白菜价”+“量身定制”！无论是卡脖子的毕设/课设/文献复现，需要灵光一现的算法改进，还是想给项目加个“外挂”，这里都有便宜又好用的代码方案等你发现！低成本，高适配，助你轻松通关！速来围观 👉
告别Print: Python调试入门，用PDB高效找Bug
1. 为什么需要专业的调试工具
1.1 Print调试的局限性
在Python开发过程中，很多开发者习惯使用print()语句进行调试，这种方法虽然简单直接，但存在诸多局限性：
# 典型的print调试示例
def calculate_stats(data):
print(f"输入数据: {data}") # 调试语句1
mean = sum(data) / len(data)
print(f"计算均值: {mean}") # 调试语句2
variance = sum((x - mean) ** 2 for x in data) / len(data)
print(f"计算方差: {variance}") # 调试语句3
std_dev = variance ** 0.5
print(f"最终结果: {std_dev}") # 调试语句4
return std_dev
# 测试代码
data = [1, 2, 3, 4, 5]
result = calculate_stats(data)
print(f"最终输出: {result}")Print调试的主要问题：
- 代码污染：调试语句与业务逻辑混杂
- 效率低下：需要频繁添加、删除print语句
- 信息有限：只能看到输出值，无法观察程序状态
- 难以调试复杂问题：对于循环、递归、异常处理等场景力不从心
1.2 专业调试工具的优势
专业的调试工具如PDB（Python Debugger）提供了以下优势：
- 断点设置：在特定位置暂停程序执行
- 变量检查：实时查看和修改变量值
- 步进执行：逐行或逐函数执行代码
- 调用栈查看：了解函数调用关系
- 条件断点：满足特定条件时暂停
- 事后调试：分析程序崩溃后的状态
2. PDB基础入门
2.1 PDB简介
PDB是Python的标准调试器，无需安装额外包，是Python标准库的一部分。它提供了交互式的调试环境，可以帮助开发者深入理解代码执行过程。
2.2 启动PDB的三种方式
2.2.1 命令行直接启动
# 直接调试Python脚本
python -m pdb script.py
# 示例：调试一个简单的计算器程序
python -m pdb calculator.py2.2.2 在代码中嵌入断点
# 方法1：使用pdb.set_trace()
import pdb
def faulty_function(x, y):
result = x + y
pdb.set_trace() # 在此处进入调试器
result *= 2
return result
# 方法2：Python 3.7+ 使用breakpoint()
def modern_function(x, y):
result = x + y
breakpoint() # 等同于pdb.set_trace()，但更现代
result *= 2
return result2.2.3 事后调试
import pdb
import traceback
def problematic_function():
try:
# 可能出错的代码
x = 1 / 0
except Exception as e:
print("发生异常，进入调试模式:")
traceback.print_exc()
pdb.post_mortem() # 事后调试，检查异常发生时的状态
problematic_function()3. PDB核心命令详解
3.1 基本调试命令
# 演示用的示例代码：complex_calculation.py
def factorial(n):
"""计算阶乘"""
if n <= 1:
return 1
else:
return n * factorial(n - 1)
def calculate_combinations(n, k):
"""计算组合数C(n, k)"""
# 设置断点
import pdb; pdb.set_trace()
if k < 0 or k > n:
return 0
numerator = factorial(n)
denominator = factorial(k) * factorial(n - k)
result = numerator // denominator
return result
# 测试
if __name__ == "__main__":
print("计算C(5, 2):")
result = calculate_combinations(5, 2)
print(f"结果: {result}")常用PDB命令表：
| 命令 | 缩写 | 功能描述 |
|---|---|---|
help | h | 查看帮助信息 |
next | n | 执行下一行 |
step | s | 进入函数调用 |
continue | c | 继续执行直到下一个断点 |
print | p | 打印变量值 |
list | l | 显示当前代码位置 |
return | r | 执行直到当前函数返回 |
where | w | 显示调用栈 |
quit | q | 退出调试器 |
3.2 高级调试技巧
3.2.1 条件断点
def process_data(data_list):
"""处理数据列表，在特定条件下设置断点"""
results = []
for i, item in enumerate(data_list):
# 条件断点：当i等于3时暂停
if i == 3:
import pdb; pdb.set_trace()
# 处理数据
processed = item * 2 + 10
# 另一个条件：当处理结果大于50时暂停
if processed > 50:
import pdb; pdb.set_trace()
results.append(processed)
return results
# 更优雅的条件断点设置
def smart_breakpoint(condition, *args):
"""智能断点函数"""
if condition:
breakpoint()
def improved_processor(data_list):
results = []
for i, item in enumerate(data_list):
# 使用智能断点
smart_breakpoint(i == 2, f"索引i={i}")
smart_breakpoint(item > 100, f"数据项={item}")
results.append(item ** 2)
return results3.2.2 观察点（Watchpoint）
class DataProcessor:
def __init__(self):
self.counter = 0
self.data = []
def add_data(self, value):
"""添加数据并计数"""
self.data.append(value)
self.counter += 1
# 当counter达到特定值时调试
if self.counter == 5:
breakpoint() # 观察计数器变化
def process(self):
"""处理数据"""
total = sum(self.data)
# 观察总和变化
if total > 100:
breakpoint()
return total
# 使用示例
processor = DataProcessor()
for i in range(10):
processor.add_data(i * 10)
result = processor.process()4. 实战：调试复杂程序
4.1 调试递归函数
# recursive_debug.py
def fibonacci(n, depth=0):
"""计算斐波那契数列（带调试信息）"""
indent = " " * depth
print(f"{indent}fibonacci({n})被调用")
# 设置条件断点：在特定深度或数值时暂停
if depth == 3 or n == 2:
breakpoint()
if n <= 1:
print(f"{indent}返回: {n}")
return n
else:
result = fibonacci(n-1, depth+1) + fibonacci(n-2, depth+1)
print(f"{indent}返回: {result}")
return result
def debug_fibonacci():
"""专门用于调试的包装函数"""
print("开始计算斐波那契数列:")
result = fibonacci(5)
print(f"最终结果: {result}")
return result
if __name__ == "__main__":
debug_fibonacci()4.2 调试面向对象程序
# oop_debug.py
class BankAccount:
def __init__(self, account_holder, initial_balance=0):
self.account_holder = account_holder
self._balance = initial_balance
self.transaction_history = []
print(f"账户创建: {account_holder}, 初始余额: {initial_balance}")
def deposit(self, amount):
"""存款"""
if amount <= 0:
raise ValueError("存款金额必须大于0")
self._balance += amount
self.transaction_history.append(('deposit', amount))
# 调试点：大额存款时暂停
if amount > 1000:
breakpoint()
print(f"存款: {amount}, 新余额: {self._balance}")
return self._balance
def withdraw(self, amount):
"""取款"""
if amount <= 0:
raise ValueError("取款金额必须大于0")
if amount > self._balance:
raise ValueError("余额不足")
self._balance -= amount
self.transaction_history.append(('withdraw', amount))
# 调试点：余额低于阈值时暂停
if self._balance < 100:
breakpoint()
print(f"取款: {amount}, 新余额: {self._balance}")
return self._balance
def transfer(self, amount, target_account):
"""转账"""
print(f"开始转账: {amount} 到 {target_account.account_holder}")
# 调试点：转账操作
breakpoint()
self.withdraw(amount)
target_account.deposit(amount)
self.transaction_history.append(('transfer', amount, target_account.account_holder))
print("转账完成")
@property
def balance(self):
return self._balance
def test_bank_account():
"""测试银行账户功能"""
# 创建账户
alice = BankAccount("Alice", 1000)
bob = BankAccount("Bob", 500)
# 执行操作
alice.deposit(200)
alice.withdraw(100)
alice.transfer(300, bob)
# 检查最终状态
print(f"Alice余额: {alice.balance}")
print(f"Bob余额: {bob.balance}")
print(f"Alice交易记录: {alice.transaction_history}")
if __name__ == "__main__":
test_bank_account()5. 高级PDB技巧
5.1 PDB配置与自定义
# .pdbrc.py - PDB启动配置文件
import pdb
def setup_pdb():
"""PDB配置函数"""
# 自定义显示格式
pdb.DefaultConfig.stdin = sys.stdin
pdb.DefaultConfig.stdout = sys.stdout
pdb.DefaultConfig.use_terminal = True
# 自定义提示符
pdb.DefaultConfig.prompt = "(我的调试器) "
# 自动导入常用模块
import os
import sys
print("PDB调试环境已初始化")
# 常用的调试辅助函数
def debug_vars(local_vars):
"""调试变量显示函数"""
print("=== 变量状态 ===")
for name, value in local_vars.items():
if not name.startswith('_'):
print(f"{name}: {value}")
def whereami():
"""显示当前位置"""
import inspect
frame = inspect.currentframe().f_back
print(f"文件: {frame.f_code.co_filename}")
print(f"行号: {frame.f_lineno}")
print(f"函数: {frame.f_code.co_name}")
# 在PDB中使用的便捷命令
class DebugCommands:
@staticmethod
def show_locals():
"""显示所有局部变量"""
import inspect
frame = inspect.currentframe().f_back.f_back
locals_dict = frame.f_locals
debug_vars(locals_dict)
@staticmethod
def show_globals():
"""显示所有全局变量"""
import inspect
frame = inspect.currentframe().f_back.f_back
globals_dict = frame.f_globals
debug_vars(globals_dict)5.2 远程调试
# remote_debug.py
import pdb
import socket
import threading
class RemoteDebugger:
"""远程调试器类"""
def __init__(self, host='localhost', port=4444):
self.host = host
self.port = port
self.debugger = None
def start_remote_debug(self):
"""启动远程调试服务"""
def debug_server():
# 创建socket服务器
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((self.host, self.port))
server.listen(1)
print(f"远程调试服务启动在 {self.host}:{self.port}")
print("等待调试客户端连接...")
client, addr = server.accept()
print(f"调试客户端已连接: {addr}")
# 重定向输入输出到socket
sys.stdin = client.makefile('r')
sys.stdout = client.makefile('w')
sys.stderr = client.makefile('w')
# 启动PDB
self.debugger = pdb.Pdb(stdin=sys.stdin, stdout=sys.stdout)
self.debugger.set_trace()
# 在新线程中运行调试服务器
debug_thread = threading.Thread(target=debug_server)
debug_thread.daemon = True
debug_thread.start()
def set_breakpoint(self, frame=None):
"""设置远程断点"""
if self.debugger:
self.debugger.set_trace(frame)
# 使用示例
def complex_algorithm(data):
"""复杂的算法函数"""
results = []
# 设置远程调试断点
debugger = RemoteDebugger()
debugger.start_remote_debug()
for i, item in enumerate(data):
# 业务逻辑
processed = item ** 2 + item * 2 + 1
# 在特定条件下触发远程调试
if processed > 1000:
debugger.set_breakpoint()
results.append(processed)
return results6. PDB与其他工具集成
6.1 与IDE调试器对比
# ide_integration.py
"""
演示如何在不同IDE中使用调试功能
"""
def compare_debugging_methods():
"""比较不同调试方法"""
methods = {
"PDB": {
"优点": ["无需IDE", "服务器环境可用", "灵活性强"],
"缺点": ["命令行操作", "学习曲线较陡"],
"适用场景": ["生产环境调试", "远程调试", "简单脚本"]
},
"PyCharm调试器": {
"优点": ["图形化界面", "功能丰富", "易用性好"],
"缺点": ["需要IDE", "资源占用较多"],
"适用场景": ["开发环境", "复杂项目", "团队协作"]
},
"VSCode调试器": {
"优点": ["轻量级", "配置灵活", "扩展性强"],
"缺点": ["需要配置", "功能相对简单"],
"适用场景": ["跨平台开发", "轻量级项目"]
}
}
return methods
def integrated_debugging_example():
"""集成调试示例"""
data = [1, 2, 3, 4, 5]
results = []
try:
for i, item in enumerate(data):
# 复杂的处理逻辑
result = item ** 2
# 条件调试：根据不同环境选择调试方法
if os.getenv('DEBUG_METHOD') == 'PDB':
import pdb; pdb.set_trace()
elif os.getenv('DEBUG_METHOD') == 'IDE':
# 这里可以设置IDE的断点
pass
results.append(result)
except Exception as e:
# 异常处理中的调试
print(f"发生异常: {e}")
if os.getenv('POST_MORTEM_DEBUG', 'false').lower() == 'true':
import pdb; pdb.post_mortem()
return results6.2 与测试框架结合
# test_debug_integration.py
import unittest
import pdb
import sys
class TestDebugIntegration(unittest.TestCase):
"""测试与调试的集成"""
def setUp(self):
"""测试准备"""
self.test_data = [1, 2, 3, 4, 5]
def debug_test_failure(self, func, *args):
"""调试测试失败的函数"""
try:
result = func(*args)
return result
except Exception as e:
print(f"测试失败，进入调试模式: {e}")
tb = sys.exc_info()[2]
pdb.post_mortem(tb)
raise
def test_complex_calculation(self):
"""测试复杂计算"""
def complex_calc(data):
results = []
for item in data:
# 可能出错的复杂计算
if item == 0:
raise ValueError("遇到零值")
result = 100 / item + item ** 2
results.append(result)
return results
# 使用调试包装器
result = self.debug_test_failure(complex_calc, self.test_data)
self.assertEqual(len(result), 5)
def test_with_breakpoint(self):
"""带断点的测试"""
def problematic_function(x):
# 设置条件断点
if x > 10:
breakpoint()
return x * 2
# 这个测试会在特定条件下触发调试器
result = problematic_function(15)
self.assertEqual(result, 30)
def run_tests_with_debug():
"""带调试的测试运行器"""
# 创建测试套件
loader = unittest.TestLoader()
suite = loader.loadTestsFromTestCase(TestDebugIntegration)
# 运行测试
runner = unittest.TextTestRunner(verbosity=2)
try:
result = runner.run(suite)
return result.wasSuccessful()
except Exception as e:
print(f"测试运行异常: {e}")
return False
if __name__ == "__main__":
run_tests_with_debug()7. 完整调试示例项目
# complete_debug_example.py
"""
完整的调试示例：一个数据处理管道
"""
import pdb
import logging
from typing import List, Dict, Any
import json
# 配置日志
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
class DataProcessingPipeline:
"""数据处理管道类"""
def __init__(self):
self.stages = []
self.context = {}
self.debug_enabled = False
def add_stage(self, stage_func, name=None):
"""添加处理阶段"""
stage_info = {
'func': stage_func,
'name': name or stage_func.__name__
}
self.stages.append(stage_info)
logger.debug(f"添加处理阶段: {stage_info['name']}")
def enable_debug(self, enabled=True):
"""启用调试模式"""
self.debug_enabled = enabled
if enabled:
logger.info("调试模式已启用")
def execute(self, input_data: List[Dict]) -> List[Dict]:
"""执行数据处理管道"""
logger.info(f"开始处理数据，共{len(input_data)}条记录")
current_data = input_data.copy()
for i, stage in enumerate(self.stages):
stage_name = stage['name']
stage_func = stage['func']
logger.info(f"执行阶段 {i+1}/{len(self.stages)}: {stage_name}")
try:
# 调试点：每个阶段开始前
if self.debug_enabled:
logger.debug(f"阶段开始前调试点")
self._debug_checkpoint(f"before_stage_{i}", current_data)
# 执行处理阶段
current_data = stage_func(current_data, self.context)
# 调试点：每个阶段结束后
if self.debug_enabled:
logger.debug(f"阶段结束后调试点")
self._debug_checkpoint(f"after_stage_{i}", current_data)
logger.info(f"阶段完成: {stage_name}")
except Exception as e:
logger.error(f"阶段执行失败: {stage_name}, 错误: {e}")
logger.info("进入调试模式...")
pdb.post_mortem()
raise
logger.info("数据处理管道执行完成")
return current_data
def _debug_checkpoint(self, checkpoint_name, data):
"""调试检查点"""
print(f"\n=== 调试检查点: {checkpoint_name} ===")
print(f"数据条数: {len(data)}")
if data:
sample = data[0] if isinstance(data, list) else data
print(f"数据样本: {json.dumps(sample, indent=2, ensure_ascii=False)}")
# 交互式调试
if self.debug_enabled:
response = input("继续执行? (y/n/debug): ").lower()
if response == 'n':
raise KeyboardInterrupt("用户中断执行")
elif response == 'debug':
breakpoint()
# 具体的处理阶段函数
def stage_clean_data(data, context):
"""数据清洗阶段"""
logger.debug("开始数据清洗")
cleaned_data = []
for record in data:
# 简单的数据清洗
cleaned_record = {k: v for k, v in record.items() if v is not None}
# 调试点：检查特定条件
if len(cleaned_record) == 0:
logger.warning("发现空记录")
if context.get('debug_empty_records', False):
breakpoint()
cleaned_data.append(cleaned_record)
return cleaned_data
def stage_calculate_stats(data, context):
"""统计计算阶段"""
logger.debug("开始统计计算")
if not data:
return data
# 计算基本统计信息
numeric_fields = []
for record in data:
for key, value in record.items():
if isinstance(value, (int, float)) and key not in numeric_fields:
numeric_fields.append(key)
# 为每个数值字段计算统计量
stats = {}
for field in numeric_fields:
values = [record.get(field, 0) for record in data if field in record]
if values:
stats[field] = {
'count': len(values),
'mean': sum(values) / len(values),
'min': min(values),
'max': max(values)
}
context['statistics'] = stats
logger.debug(f"计算统计信息: {stats}")
return data
def stage_enrich_data(data, context):
"""数据增强阶段"""
logger.debug("开始数据增强")
enriched_data = []
for i, record in enumerate(data):
# 添加索引信息
record['_index'] = i
record['_processed'] = True
# 调试点：处理特定记录时暂停
if i == len(data) // 2: # 处理到中间记录时
logger.debug(f"处理中间记录: {i}")
if context.get('debug_midpoint', False):
breakpoint()
enriched_data.append(record)
return enriched_data
def main():
"""主函数：演示完整的数据处理管道"""
# 创建测试数据
sample_data = [
{'name': 'Alice', 'age': 25, 'score': 85.5},
{'name': 'Bob', 'age': 30, 'score': 92.0},
{'name': 'Charlie', 'age': None, 'score': 78.5}, # 包含空值
{'name': 'Diana', 'age': 28, 'score': None}, # 包含空值
{'name': 'Eve', 'age': 35, 'score': 88.0}
]
# 创建处理管道
pipeline = DataProcessingPipeline()
pipeline.enable_debug(True) # 启用调试模式
# 配置调试上下文
pipeline.context = {
'debug_empty_records': True,
'debug_midpoint': True
}
# 添加处理阶段
pipeline.add_stage(stage_clean_data, "数据清洗")
pipeline.add_stage(stage_calculate_stats, "统计计算")
pipeline.add_stage(stage_enrich_data, "数据增强")
try:
# 执行管道
result = pipeline.execute(sample_data)
# 输出结果
print("\n=== 处理结果 ===")
for record in result:
print(json.dumps(record, indent=2, ensure_ascii=False))
print(f"\n=== 统计信息 ===")
print(json.dumps(pipeline.context['statistics'], indent=2, ensure_ascii=False))
except Exception as e:
logger.error(f"管道执行失败: {e}")
print("进入事后调试...")
pdb.post_mortem()
if __name__ == "__main__":
main()8. 调试最佳实践与工作流
8.1 系统化的调试方法
# debugging_workflow.py
"""
系统化的调试工作流实现
"""
class SystematicDebugger:
"""系统化调试器"""
def __init__(self):
self.breakpoints = {}
self.watchpoints = {}
self.history = []
def add_breakpoint(self, condition_func, description=""):
"""添加断点"""
bp_id = len(self.breakpoints) + 1
self.breakpoints[bp_id] = {
'condition': condition_func,
'description': description,
'hit_count': 0
}
return bp_id
def add_watchpoint(self, variable_name, condition_func, description=""):
"""添加观察点"""
wp_id = len(self.watchpoints) + 1
self.watchpoints[wp_id] = {
'variable': variable_name,
'condition': condition_func,
'description': description,
'hit_count': 0
}
return wp_id
def check_breakpoints(self, local_vars, global_vars):
"""检查断点条件"""
for bp_id, bp_info in self.breakpoints.items():
try:
if bp_info['condition'](local_vars, global_vars):
bp_info['hit_count'] += 1
self.history.append({
'type': 'breakpoint',
'id': bp_id,
'description': bp_info['description'],
'timestamp': self._get_timestamp(),
'variables': local_vars.copy()
})
print(f"\n*** 断点命中: {bp_info['description']} ***")
print(f"命中次数: {bp_info['hit_count']}")
breakpoint()
except Exception as e:
print(f"断点检查错误: {e}")
def debug_function(self, func):
"""调试装饰器"""
def wrapper(*args, **kwargs):
# 记录函数调用
call_info = {
'function': func.__name__,
'args': args,
'kwargs': kwargs,
'timestamp': self._get_timestamp()
}
self.history.append(call_info)
# 执行前检查断点
frame = sys._getframe()
self.check_breakpoints(frame.f_locals, frame.f_globals)
try:
result = func(*args, **kwargs)
return result
except Exception as e:
# 异常处理
error_info = {
'function': func.__name__,
'exception': str(e),
'timestamp': self._get_timestamp()
}
self.history.append(error_info)
print(f"函数 {func.__name__} 执行异常: {e}")
pdb.post_mortem()
raise
return wrapper
def _get_timestamp(self):
"""获取时间戳"""
from datetime import datetime
return datetime.now().isoformat()
def print_debug_report(self):
"""打印调试报告"""
print("\n" + "="*50)
print("调试报告")
print("="*50)
print(f"\n断点统计:")
for bp_id, bp_info in self.breakpoints.items():
print(f" {bp_id}: {bp_info['description']} - 命中 {bp_info['hit_count']} 次")
print(f"\n调用历史 (最近10次):")
for i, event in enumerate(self.history[-10:]):
print(f" {i+1}. {event}")
# 使用示例
def demo_systematic_debugging():
"""演示系统化调试"""
debugger = SystematicDebugger()
# 定义断点条件
def bp_high_value(locals, globals):
return 'result' in locals and locals['result'] > 100
def bp_negative_input(locals, globals):
return 'x' in locals and locals['x'] < 0
# 添加断点
debugger.add_breakpoint(bp_high_value, "结果值大于100")
debugger.add_breakpoint(bp_negative_input, "输入值为负数")
# 调试函数
@debugger.debug_function
def complex_calculation(x, y):
result = x * y + (x + y) ** 2
debugger.check_breakpoints(locals(), globals())
return result
# 测试调用
test_cases = [(2, 3), (10, 5), (-1, 10), (8, 12)]
for x, y in test_cases:
print(f"\n计算: complex_calculation({x}, {y})")
try:
result = complex_calculation(x, y)
print(f"结果: {result}")
except Exception as e:
print(f"错误: {e}")
# 生成调试报告
debugger.print_debug_report()
if __name__ == "__main__":
demo_systematic_debugging()9. 调试性能优化
9.1 生产环境调试策略
# production_debug.py
"""
生产环境调试策略
"""
import logging
import time
from functools import wraps
def production_debug(enable_tracing=False, slow_threshold=1.0):
"""生产环境调试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
logger = logging.getLogger(func.__module__)
# 记录函数开始
if enable_tracing:
logger.debug(f"函数开始: {func.__name__}")
try:
result = func(*args, **kwargs)
execution_time = time.time() - start_time
# 检查执行时间
if execution_time > slow_threshold:
logger.warning(
f"函数执行缓慢: {func.__name__}, "
f"耗时: {execution_time:.2f}秒"
)
if enable_tracing:
logger.debug(f"函数完成: {func.__name__}")
return result
except Exception as e:
execution_time = time.time() - start_time
logger.error(
f"函数执行错误: {func.__name__}, "
f"错误: {e}, 耗时: {execution_time:.2f}秒"
)
# 生产环境下的安全调试
if os.getenv('ENABLE_PRODUCTION_DEBUG', 'false').lower() == 'true':
logger.info("进入生产环境调试模式")
import pdb
pdb.post_mortem()
raise
return wrapper
return decorator
class PerformanceMonitor:
"""性能监控器"""
def __init__(self):
self.metrics = {}
self.logger = logging.getLogger('performance')
def track_performance(self, operation_name):
"""性能跟踪装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
start_memory = self._get_memory_usage()
try:
result = func(*args, **kwargs)
return result
finally:
end_time = time.time()
end_memory = self._get_memory_usage()
execution_time = end_time - start_time
memory_used = end_memory - start_memory
# 记录指标
self._record_metrics(
operation_name, execution_time, memory_used
)
# 检查性能问题
self._check_performance_issues(
operation_name, execution_time, memory_used
)
return wrapper
return decorator
def _get_memory_usage(self):
"""获取内存使用量"""
import psutil
process = psutil.Process()
return process.memory_info().rss / 1024 / 1024 # MB
def _record_metrics(self, operation, time_used, memory_used):
"""记录性能指标"""
if operation not in self.metrics:
self.metrics[operation] = {
'count': 0,
'total_time': 0,
'max_time': 0,
'total_memory': 0,
'max_memory': 0
}
metrics = self.metrics[operation]
metrics['count'] += 1
metrics['total_time'] += time_used
metrics['max_time'] = max(metrics['max_time'], time_used)
metrics['total_memory'] += memory_used
metrics['max_memory'] = max(metrics['max_memory'], memory_used)
def _check_performance_issues(self, operation, time_used, memory_used):
"""检查性能问题"""
# 时间阈值检查
if time_used > 5.0: # 5秒阈值
self.logger.warning(
f"操作 {operation} 执行时间过长: {time_used:.2f}秒"
)
# 内存阈值检查
if memory_used > 100: # 100MB阈值
self.logger.warning(
f"操作 {operation} 内存使用过多: {memory_used:.2f}MB"
)
def generate_report(self):
"""生成性能报告"""
report = ["性能监控报告:", "="*50]
for operation, metrics in self.metrics.items():
avg_time = metrics['total_time'] / metrics['count']
avg_memory = metrics['total_memory'] / metrics['count']
report.extend([
f"操作: {operation}",
f" 调用次数: {metrics['count']}",
f" 平均时间: {avg_time:.3f}秒",
f" 最大时间: {metrics['max_time']:.3f}秒",
f" 平均内存: {avg_memory:.2f}MB",
f" 最大内存: {metrics['max_memory']:.2f}MB",
""
])
return "\n".join(report)
# 使用示例
def demo_production_debugging():
"""演示生产环境调试"""
monitor = PerformanceMonitor()
@monitor.track_performance("数据计算")
@production_debug(enable_tracing=True, slow_threshold=0.5)
def heavy_calculation(n):
"""耗时的计算函数"""
result = 0
for i in range(n):
result += i ** 2
time.sleep(0.01) # 模拟耗时操作
return result
# 测试调用
for i in [100, 200, 300]:
print(f"计算 n={i}")
result = heavy_calculation(i)
print(f"结果: {result}")
# 生成报告
print(monitor.generate_report())
if __name__ == "__main__":
demo_production_debugging()10. 总结
通过本文的全面介绍，我们深入探讨了Python PDB调试器的强大功能和使用技巧。从基础的断点设置到高级的远程调试，从简单的脚本调试到复杂的生产环境调试，PDB为Python开发者提供了全面的调试