5.4 小结和复习

本章学到了什么？

小白版总结

这一章我们学会了让程序"更健壮"——不怕出错，出错也能优雅处理：

第1节：异常处理 → 程序出错时，不崩溃，有备案
第2节：日志系统 → 程序运行时，记日记，方便查
第3节：调试技巧 → 程序有bug，会查找，能修复

一图看懂三者关系：

          程序运行
             │
    ┌────────┼────────┐
    ▼        ▼        ▼
 正常运行  出现错误  需要调试
    │        │        │
    ▼        ▼        ▼
  日志记录  异常处理  调试工具
  (logging) (try/except) (pdb)

本章核心概念

1. 异常处理

小白速记：

python

try:      # 尝试执行
except:   # 出错了怎么办
finally:  # 不管成功失败都要做的事
raise:    # 主动抛出错误

try/except/finally：捕获和处理异常
自定义异常：创建特定的错误类型
异常链：使用 raise ... from ...
最佳实践：具体的异常优先于通用异常

2. 日志系统

小白速记：

python

import logging
logging.debug("调试信息")    # 开发时看
logging.info("普通信息")     # 日常运行看
logging.warning("警告")      # 可能有问题
logging.error("错误")        # 确实出错了
logging.critical("严重错误") # 系统要崩了

logging 模块：Python 标准日志库
日志级别：DEBUG < INFO < WARNING < ERROR < CRITICAL
处理器：FileHandler, StreamHandler, RotatingFileHandler
格式化：自定义日志输出格式

3. 调试技巧

小白速记：
方法一句话何时用
print() 看变量值快速检查
breakpoint() 暂停程序深入排查
logging 记录日志长期监控
VS Code 图形化调试复杂问题

方法	一句话	何时用
`print()`	看变量值	快速检查
`breakpoint()`	暂停程序	深入排查
`logging`	记录日志	长期监控
VS Code	图形化调试	复杂问题

print 调试：快速简单的调试方法
pdb/breakpoint()：交互式调试器
装饰器：创建可复用的调试工具
性能分析：cProfile, tracemalloc

知识自测

小白提示：做完题再看答案，检验一下自己掌握了多少！

选择题

下面哪种异常处理方式最好？
- A. except Exception:
- B. except ValueError:
- C. except:
- D. 不处理异常
日志级别从低到高的顺序是？
- A. INFO < DEBUG < WARNING < ERROR
- B. DEBUG < INFO < WARNING < ERROR
- C. WARNING < DEBUG < INFO < ERROR
- D. DEBUG < WARNING < INFO < ERROR
breakpoint() 函数的作用是？
- A. 抛出异常
- B. 暂停程序进入调试器
- C. 设置断点但不停止
- D. 打印调试信息

查看答案

B - 应该捕获具体的异常类型，避免捕获所有异常
B - DEBUG < INFO < WARNING < ERROR < CRITICAL
B - breakpoint() 会暂停程序并进入交互式调试器

高难度编程挑战

小白注意：这个挑战难度很高，适合学完整个课程后来挑战！如果觉得太难，可以先跳过，等学完后面的课程再回来做。

挑战：智能错误恢复系统（难度：⭐⭐⭐⭐⭐）

这个系统能做什么？
想象一个"自动修复机器人"：
程序出错 → 自动分析是什么错误
可以重试的错误 → 自动重试几次
重试不行 → 使用备用方案
记录所有错误 → 方便后续分析
类比：像手机信号不好时：
先重连几次（重试）
重连不上就用 WiFi（降级）
记录断网时间（日志）

需求：设计一个智能错误恢复系统，能够自动从各种错误中恢复并继续执行。

核心功能：

错误分类：将错误分为可恢复和不可恢复
自动重试：对可恢复错误自动重试
降级策略：主方法失败时使用备用方法
错误聚合：收集和分析错误模式
智能决策：根据错误历史调整重试策略
监控告警：错误率超过阈值时告警

代码框架：

python

from typing import Callable, Any, Optional, Dict, List, Type
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import time
import statistics
from collections import deque, defaultdict

class ErrorSeverity(Enum):
    """错误严重程度"""
    LOW = "low"          # 可忽略
    MEDIUM = "medium"    # 可重试
    HIGH = "high"        # 需要降级
    CRITICAL = "critical"  # 不可恢复

@dataclass
class ErrorRecord:
    """错误记录"""
    timestamp: datetime
    exception_type: str
    exception_message: str
    severity: ErrorSeverity
    context: Dict[str, Any]
    recovered: bool = False
    retry_count: int = 0

class ErrorClassifier:
    """错误分类器"""
    
    def __init__(self):
        # 定义错误分类规则
        self.rules: Dict[Type[Exception], ErrorSeverity] = {
            ConnectionError: ErrorSeverity.MEDIUM,
            TimeoutError: ErrorSeverity.MEDIUM,
            ValueError: ErrorSeverity.HIGH,
            KeyError: ErrorSeverity.HIGH,
            MemoryError: ErrorSeverity.CRITICAL,
            KeyboardInterrupt: ErrorSeverity.CRITICAL,
        }
    
    def classify(self, exception: Exception) -> ErrorSeverity:
        """分类异常"""
        # TODO: 实现错误分类逻辑
        # - 检查异常类型
        # - 检查异常消息
        # - 返回严重程度
        pass

class RecoveryStrategy:
    """恢复策略"""
    
    def __init__(self):
        self.strategies: Dict[ErrorSeverity, Callable] = {}
    
    def register(
        self,
        severity: ErrorSeverity,
        strategy: Callable[[Exception, Dict[str, Any]], Any]
    ):
        """注册恢复策略"""
        self.strategies[severity] = strategy
    
    def recover(
        self,
        severity: ErrorSeverity,
        exception: Exception,
        context: Dict[str, Any]
    ) -> Any:
        """执行恢复"""
        # TODO: 实现恢复逻辑
        pass

class ErrorAnalyzer:
    """错误分析器"""
    
    def __init__(self, window_size: int = 100):
        self.window_size = window_size
        self.recent_errors: deque = deque(maxlen=window_size)
        self.error_patterns: Dict[str, List[ErrorRecord]] = defaultdict(list)
    
    def record(self, error: ErrorRecord):
        """记录错误"""
        # TODO: 实现错误记录
        pass
    
    def analyze_pattern(self) -> Dict[str, Any]:
        """分析错误模式"""
        # TODO: 实现模式分析
        # - 错误频率
        # - 错误分布
        # - 恢复成功率
        # - 建议的重试参数
        pass
    
    def get_error_rate(self, time_window: timedelta) -> float:
        """计算错误率"""
        # TODO: 计算指定时间窗口内的错误率
        pass
    
    def should_alert(self) -> bool:
        """是否应该告警"""
        # TODO: 检查是否满足告警条件
        pass

class SmartRecovery:
    """智能错误恢复系统"""
    
    def __init__(
        self,
        max_retries: int = 3,
        initial_delay: float = 1.0,
        backoff_factor: float = 2.0,
        alert_threshold: float = 0.1  # 10% 错误率
    ):
        self.max_retries = max_retries
        self.initial_delay = initial_delay
        self.backoff_factor = backoff_factor
        self.alert_threshold = alert_threshold
        
        self.classifier = ErrorClassifier()
        self.recovery = RecoveryStrategy()
        self.analyzer = ErrorAnalyzer()
        
        self._setup_default_strategies()
    
    def _setup_default_strategies(self):
        """设置默认恢复策略"""
        # TODO: 注册各个严重程度的恢复策略
        pass
    
    def execute_with_recovery(
        self,
        func: Callable,
        *args,
        fallback: Optional[Callable] = None,
        context: Optional[Dict[str, Any]] = None,
        **kwargs
    ) -> Any:
        """带恢复的执行"""
        # TODO: 实现以下逻辑
        # 1. 尝试执行函数
        # 2. 捕获异常并分类
        # 3. 根据严重程度决定恢复策略
        # 4. 记录错误
        # 5. 分析错误模式
        # 6. 检查是否需要告警
        # 7. 返回结果或执行降级
        pass
    
    def retry_with_backoff(
        self,
        func: Callable,
        *args,
        **kwargs
    ) -> Any:
        """带退避的重试"""
        # TODO: 实现智能重试
        # - 根据历史数据调整重试参数
        # - 实现指数退避
        # - 记录重试次数
        pass
    
    def get_stats(self) -> Dict[str, Any]:
        """获取统计信息"""
        # TODO: 返回错误和恢复统计
        pass

# 使用示例
recovery_system = SmartRecovery()

# 定义可能失败的函数
def unstable_api_call(param: str) -> str:
    """不稳定的 API 调用"""
    import random
    if random.random() < 0.3:  # 30% 失败率
        raise ConnectionError("API 连接失败")
    return f"成功: {param}"

# 定义降级函数
def fallback_method(param: str) -> str:
    """降级方法"""
    return f"降级结果: {param}"

# 执行带恢复的调用
try:
    result = recovery_system.execute_with_recovery(
        unstable_api_call,
        "test_param",
        fallback=fallback_method,
        context={"user_id": "123"}
    )
    print(f"结果: {result}")
except Exception as e:
    print(f"最终失败: {e}")

# 查看统计
stats = recovery_system.get_stats()
print(f"\n统计信息: {stats}")

# 检查是否需要告警
if recovery_system.analyzer.should_alert():
    print("⚠️ 错误率过高，需要人工介入!")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202

评分标准：

错误分类系统（15分）
重试机制和退避策略（20分）
降级策略（20分）
错误分析和模式识别（25分）
告警系统（20分）

参考解决方案

点击查看部分参考实现

python

from typing import Callable, Any, Optional, Dict
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
import time
from collections import deque

class ErrorSeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ErrorRecord:
    timestamp: datetime
    exception_type: str
    exception_message: str
    severity: ErrorSeverity
    context: Dict[str, Any]
    recovered: bool = False
    retry_count: int = 0

class ErrorClassifier:
    def __init__(self):
        self.rules = {
            ConnectionError: ErrorSeverity.MEDIUM,
            TimeoutError: ErrorSeverity.MEDIUM,
            ValueError: ErrorSeverity.HIGH,
            MemoryError: ErrorSeverity.CRITICAL,
        }
    
    def classify(self, exception: Exception) -> ErrorSeverity:
        exc_type = type(exception)
        
        # 检查精确匹配
        if exc_type in self.rules:
            return self.rules[exc_type]
        
        # 检查父类
        for exc_class, severity in self.rules.items():
            if isinstance(exception, exc_class):
                return severity
        
        # 默认为 HIGH
        return ErrorSeverity.HIGH

class RecoveryStrategy:
    def __init__(self):
        self.strategies = {}
    
    def register(self, severity: ErrorSeverity, strategy: Callable):
        self.strategies[severity] = strategy
    
    def recover(
        self,
        severity: ErrorSeverity,
        exception: Exception,
        context: Dict[str, Any]
    ) -> Any:
        if severity in self.strategies:
            return self.strategies[severity](exception, context)
        raise exception

class SmartRecovery:
    def __init__(
        self,
        max_retries: int = 3,
        initial_delay: float = 1.0,
        backoff_factor: float = 2.0
    ):
        self.max_retries = max_retries
        self.initial_delay = initial_delay
        self.backoff_factor = backoff_factor
        
        self.classifier = ErrorClassifier()
        self.recovery = RecoveryStrategy()
        self.error_history: deque = deque(maxlen=100)
        
        self._setup_strategies()
    
    def _setup_strategies(self):
        # MEDIUM: 重试
        self.recovery.register(
            ErrorSeverity.MEDIUM,
            lambda exc, ctx: self._retry_strategy(exc, ctx)
        )
        
        # HIGH: 降级
        self.recovery.register(
            ErrorSeverity.HIGH,
            lambda exc, ctx: self._fallback_strategy(exc, ctx)
        )
    
    def _retry_strategy(self, exception: Exception, context: Dict[str, Any]):
        print(f"使用重试策略: {exception}")
        return "RETRY"
    
    def _fallback_strategy(self, exception: Exception, context: Dict[str, Any]):
        print(f"使用降级策略: {exception}")
        if "fallback" in context:
            return context["fallback"]()
        raise exception
    
    def execute_with_recovery(
        self,
        func: Callable,
        *args,
        fallback: Optional[Callable] = None,
        context: Optional[Dict[str, Any]] = None,
        **kwargs
    ) -> Any:
        context = context or {}
        if fallback:
            context["fallback"] = lambda: fallback(*args, **kwargs)
        
        last_exception = None
        
        for attempt in range(self.max_retries):
            try:
                result = func(*args, **kwargs)
                
                # 记录成功
                if last_exception:
                    self.error_history.append(ErrorRecord(
                        timestamp=datetime.now(),
                        exception_type=type(last_exception).__name__,
                        exception_message=str(last_exception),
                        severity=self.classifier.classify(last_exception),
                        context=context,
                        recovered=True,
                        retry_count=attempt
                    ))
                
                return result
            
            except Exception as e:
                last_exception = e
                severity = self.classifier.classify(e)
                
                print(f"尝试 {attempt + 1}/{self.max_retries} 失败: {e}")
                
                if severity == ErrorSeverity.CRITICAL:
                    raise
                
                if attempt < self.max_retries - 1:
                    delay = self.initial_delay * (self.backoff_factor ** attempt)
                    print(f"等待 {delay:.2f} 秒后重试...")
                    time.sleep(delay)
        
        # 所有重试失败，尝试恢复
        severity = self.classifier.classify(last_exception)
        
        # 记录错误
        self.error_history.append(ErrorRecord(
            timestamp=datetime.now(),
            exception_type=type(last_exception).__name__,
            exception_message=str(last_exception),
            severity=severity,
            context=context,
            recovered=False,
            retry_count=self.max_retries
        ))
        
        return self.recovery.recover(severity, last_exception, context)
    
    def get_stats(self) -> Dict[str, Any]:
        total = len(self.error_history)
        if total == 0:
            return {"total_errors": 0}
        
        recovered = sum(1 for e in self.error_history if e.recovered)
        
        return {
            "total_errors": total,
            "recovered": recovered,
            "recovery_rate": recovered / total if total > 0 else 0.0,
            "recent_errors": total
        }

学习资源

小白学习建议

新手路线图：

必会（先掌握）          进阶（需要时学）        高级（用到再学）
     │                      │                     │
     ▼                      ▼                     ▼
try/except            自定义异常              分布式追踪
print 调试            logging 模块            错误监控系统
基本日志              pdb 调试器              性能监控

记住：不需要一次学会所有内容，先掌握基础，其他用到时再查。

进阶主题

小白理解 - 这些是什么？
工具一句话解释何时需要
OpenTelemetry 追踪请求在多个服务间的流转微服务架构
Sentry 自动收集和报告程序错误生产环境监控
Prometheus 监控系统性能指标大规模部署
ELK Stack 收集、搜索、展示日志日志量很大时

工具	一句话解释	何时需要
OpenTelemetry	追踪请求在多个服务间的流转	微服务架构
Sentry	自动收集和报告程序错误	生产环境监控
Prometheus	监控系统性能指标	大规模部署
ELK Stack	收集、搜索、展示日志	日志量很大时

分布式追踪（OpenTelemetry）
错误监控（Sentry）
性能监控（Prometheus）
日志聚合（ELK Stack）

下一章：Module 6 - 异步编程基础

5.4 小结和复习 ​

本章学到了什么？ ​

本章核心概念 ​

1. 异常处理 ​

2. 日志系统 ​

3. 调试技巧 ​

知识自测 ​

选择题 ​

高难度编程挑战 ​

挑战：智能错误恢复系统（难度：⭐⭐⭐⭐⭐） ​

参考解决方案 ​

学习资源 ​

推荐阅读 ​

进阶主题 ​

5.4 小结和复习

本章学到了什么？

本章核心概念

1. 异常处理

2. 日志系统

3. 调试技巧

知识自测

选择题

高难度编程挑战

挑战：智能错误恢复系统（难度：⭐⭐⭐⭐⭐）

参考解决方案

学习资源

推荐阅读

进阶主题