项目里到处都是重复的日志代码,每个函数开头写 logger.info("开始..."),结尾写 logger.info("结束")。后来用装饰器统一处理,代码简洁了很多。这篇文章记录一下装饰器的实战用法。

什么是闭包

装饰器的基础是闭包。简单说,闭包就是函数里定义的函数,能记住外层函数的变量。

def outer(x):
    def inner(y):
        return x + y  # inner 记住了 x 的值
    return inner

add5 = outer(5)
print(add5(3))  # 输出 8
print(add5(10))  # 输出 15

坑在于:outer(5) 执行完后,按理说 x 应该被销毁了。但 inner 函数记住了 x 的值,所以还能用。

最简单的装饰器

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} 执行完成")
        return result
    return wrapper

@log_decorator
def add(a, b):
    return a + b

add(1, 2)
# 输出:
# 调用 add
# add 执行完成

@log_decorator 等价于 add = log_decorator(add)

带参数的装饰器

如果装饰器本身需要参数,要再包一层:

import time
from functools import wraps

def timer(threshold=None):
    def decorator(func):
        @wraps(func)  # 保留原函数的 __name__ 和 __doc__
        def wrapper(*args, **kwargs):
            start = time.time()
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            if threshold and elapsed > threshold:
                print(f"警告:{func.__name__} 耗时 {elapsed:.2f}s,超过阈值 {threshold}s")
            else:
                print(f"{func.__name__} 耗时 {elapsed:.2f}s")
            return result
        return wrapper
    return decorator

@timer(threshold=1)
def slow_function():
    time.sleep(2)

slow_function()
# 输出:警告:slow_function 耗时 2.00s,超过阈值 1s

坑在于:不加 @wraps(func),装饰后的函数 __name__ 会变成 wrapper,调试时很困惑。

实战:日志装饰器

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

def log_execution(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"开始执行 {func.__name__},参数: args={args}, kwargs={kwargs}")
        start = time.time()
        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(f"{func.__name__} 执行成功,耗时: {elapsed:.3f}s")
            return result
        except Exception as e:
            elapsed = time.time() - start
            logger.error(f"{func.__name__} 执行失败,耗时: {elapsed:.3f}s,错误: {e}")
            raise
    return wrapper

@log_execution
def process_data(data):
    time.sleep(0.5)
    return [x * 2 for x in data]

result = process_data([1, 2, 3])

输出:

2026-07-20 14:30:00 [INFO] 开始执行 process_data,参数: args=([1, 2, 3],), kwargs={}
2026-07-20 14:30:01 [INFO] process_data 执行成功,耗时: 0.501s

实战:重试装饰器

import time
from functools import wraps

def retry(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    print(f"第 {attempt + 1} 次失败:{e},{delay}s 后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_retries=3, delay=2)
def unstable_api():
    import random
    if random.random() < 0.7:
        raise ConnectionError("API 连接失败")
    return "成功"

result = unstable_api()

这个装饰器在网络请求场景很有用,自动重试失败的请求。

实战:缓存装饰器

from functools import wraps

def cache(func):
    cached = {}
    @wraps(func)
    def wrapper(*args):
        if args in cached:
            print(f"命中缓存:{args}")
            return cached[args]
        result = func(*args)
        cached[args] = result
        return result
    return wrapper

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(10))  # 计算
print(fibonacci(10))  # 命中缓存

坑在于:这个简单缓存没有过期机制,如果数据量大会占用内存。Python 3.9+ 可以用 functools.lru_cache

实战:权限检查装饰器

from functools import wraps

def require_permission(permission):
    def decorator(func):
        @wraps(func)
        def wrapper(user, *args, **kwargs):
            if permission not in user.get("permissions", []):
                raise PermissionError(f"用户 {user['name']} 没有 {permission} 权限")
            return func(user, *args, **kwargs)
        return wrapper
    return decorator

@require_permission("admin")
def delete_user(user, user_id):
    print(f"删除用户 {user_id}")

admin = {"name": "admin", "permissions": ["admin", "read"]}
user = {"name": "guest", "permissions": ["read"]}

delete_user(admin, 123)  # 正常执行
# delete_user(user, 123)  # 抛出 PermissionError

踩坑总结

  1. 装饰器本质是闭包,记住外层函数的变量
  2. @wraps(func) 保留原函数的元信息
  3. 带参数的装饰器要再包一层
  4. 装饰器顺序从下往上执行
  5. Python 3.9+ 用 functools.lru_cache 做缓存,比自己实现更可靠

print() 调试到装饰器统一处理,只需要花半小时学习一次,但能省下无数重复代码。

做嵌入式项目时,I2C 通信总是不稳定,有时候能通有时候不通。排查了半天,最后发现是开漏输出没加上拉电阻。这个问题很常见,90% 的工程师都忽略了这一点。

问题现象

// 配置 GPIO 为开漏输出
GPIO_InitTypeDef GPIO_InitStruct = {0};
GPIO_InitStruct.Pin = GPIO_PIN_6;  // SCL
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;  // 开漏输出
GPIO_InitStruct.Pull = GPIO_NOPULL;  // 没有上下拉
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOB, &GPIO_InitStruct);

I2C 通信时好时坏,有时候能通,有时候直接卡死。用逻辑分析仪看波形,发现 SCL 线有时候拉不上去。

原因分析

说白了,就是开漏输出的特性决定的。

推挽输出(Push-Pull)

  • 可以主动输出高电平和低电平
  • 内部有两个 MOS 管,一个拉高,一个拉低
  • 不需要外部上下拉电阻

开漏输出(Open-Drain)

  • 只能主动输出低电平
  • 高电平是靠外部上拉电阻拉上去的
  • 如果没有上拉电阻,高电平就是浮空状态

坑在于:开漏输出的"高电平"不是 MCU 主动输出的,而是靠外部上拉电阻拉上去的。如果没有上拉,高电平就是浮空,电平不确定。

解决方案

方案一:配置内部上拉

// 配置 GPIO 为开漏输出 + 内部上拉
GPIO_InitTypeDef GPIO_InitStruct = {0};
GPIO_InitStruct.Pin = GPIO_PIN_6;  // SCL
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;  // 开漏输出
GPIO_InitStruct.Pull = GPIO_PULLUP;  // 内部上拉
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOB, &GPIO_InitStruct);

坑在于:内部上拉电阻通常在 20kΩ-50kΩ,阻值比较大。对于 I2C 这种需要快速上升沿的场景,内部上拉可能不够。

方案二:外部上拉电阻(推荐)

// 在硬件上加外部上拉电阻
// SCL 线:4.7kΩ 上拉到 VCC
// SDA 线:4.7kΩ 上拉到 VCC

I2C 协议推荐使用 4.7kΩ 上拉电阻。如果总线上设备多或者线长,可以适当减小阻值(比如 2.2kΩ)。

方案三:根据场景选择

// 场景 1:普通 GPIO 控制 LED
// 推挽输出,不需要上下拉
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;
GPIO_InitStruct.Pull = GPIO_NOPULL;

// 场景 2:I2C 通信
// 开漏输出 + 外部上拉
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;
GPIO_InitStruct.Pull = GPIO_NOPULL;  // 用外部上拉

// 场景 3:按键输入
// 输入模式 + 内部上拉
GPIO_InitStruct.Mode = GPIO_MODE_INPUT;
GPIO_InitStruct.Pull = GPIO_PULLUP;

// 场景 4:UART TX
// 推挽输出,空闲状态保持高电平
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;
GPIO_InitStruct.Pull = GPIO_NOPULL;

常见场景配置

I2C 通信

// I2C GPIO 配置
GPIO_InitStruct.Pin = GPIO_PIN_6 | GPIO_PIN_7;  // SCL, SDA
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;  // 开漏输出
GPIO_InitStruct.Pull = GPIO_PULLUP;  // 内部上拉(建议用外部 4.7kΩ)
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOB, &GPIO_InitStruct);

SPI 通信

// SPI GPIO 配置
GPIO_InitStruct.Pin = GPIO_PIN_5 | GPIO_PIN_7;  // SCK, MOSI
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;  // 推挽输出
GPIO_InitStruct.Pull = GPIO_NOPULL;
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOA, &GPIO_InitStruct);

UART 通信

// UART TX 配置
GPIO_InitStruct.Pin = GPIO_PIN_9;  // TX
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;  // 推挽输出
GPIO_InitStruct.Pull = GPIO_PULLUP;  // 空闲状态高电平
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOA, &GPIO_InitStruct);

踩坑总结

  1. 开漏输出只能主动拉低,高电平靠外部上拉
  2. I2C 必须用开漏输出 + 上拉电阻
  3. SPI、UART 用推挽输出,不需要上下拉
  4. 内部上拉电阻阻值大,高速通信建议用外部上拉
  5. 记住口诀:"推挽上下拉是白搭,开漏不拉就抓瞎"

这个问题的坑在于:开漏输出的高电平是靠上拉电阻实现的,没有上拉就是浮空。理解了这个原理,配置就很清晰了。

凌晨 2 点收到告警,服务挂了。SSH 上去一看,No space left on device。df 一看,根分区 100% 了。但 du 一算,明明只用了 20G,分区有 40G。坑在于:有已删除的文件还占着空间。

问题现象

df -h
# Filesystem      Size  Used Avail Use% Mounted on
# /dev/vda1        40G   40G     0 100% /

du -sh /*
# 加起来只有 20G

df 说用了 40G,du 说只用了 20G。差的 20G 去哪了?

原因分析

说白了,就是有文件被删除了,但进程还持有文件句柄。Linux 的机制是:文件被删除后,只要还有进程在用它,磁盘空间就不会释放。

坑在于:du 只统计当前存在的文件,不统计已删除但还被占用的文件。

排查过程

第一步:找出已删除但未释放的文件

# 查找被删除但还被占用的文件
lsof +L1

# 输出类似:
# COMMAND    PID USER   FD   TYPE DEVICE SIZE/NODE NODE NAME
# java     12345 root    1w  REG  253,1 21474836480  1234567 /var/log/app.log (deleted)

lsof +L1 查找所有已删除但还被占用的文件(links < 1)。输出显示 java 进程占用了一个 20G 的已删除文件。

第二步:确认文件大小

# 查看被删除文件的大小
ls -lh /proc/12345/fd/1

# 或者
ls -l /proc/12345/fd/ | grep deleted

# 输出:
# l-wx------ 1 root root 64 Jul 20 02:00 1 -> /var/log/app.log (deleted)

确认了,/var/log/app.log 被删除了,但 java 进程还持有它的句柄,占着 20G 空间。

第三步:释放空间

有两个方案:

方案一:重启进程(推荐)

# 重启 java 进程,释放文件句柄
systemctl restart java-app

重启后,已删除的文件句柄被释放,磁盘空间回来了。

方案二:不重启,清空文件内容

# 找到被删除文件的文件描述符
ls -l /proc/12345/fd/ | grep deleted
# 输出:1 -> /var/log/app.log (deleted)

# 清空文件内容(不删除文件,只清空)
> /proc/12345/fd/1

这个方法不需要重启进程,但坑在于:只能清空当前内容,进程继续写入还是会占用空间。

验证修复

# 重启后检查
df -h
# Filesystem      Size  Used Avail Use% Mounted on
# /dev/vda1        40G   20G   18G  53% /

# 确认没有已删除未释放的文件
lsof +L1
# 无输出

空间回来了,从 100% 降到 53%。

预防措施

1. 配置日志轮转

# /etc/logrotate.d/app
/var/log/app.log {
    daily
    rotate 7
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
}

copytruncate 会先复制再清空,避免进程持有已删除文件的句柄。

2. 监控磁盘使用率

# 添加 crontab 定期检查
*/5 * * * * df -h / | awk 'NR==2{print $5}' | sed 's/%//' | awk '$1>80{print "磁盘使用率超过80%: "$1"%"}' | mail -s "磁盘告警" admin@example.com

3. 清理大文件

# 找出大于 100M 的文件
find / -type f -size +100M -exec ls -lh {} \; 2>/dev/null

# 找出大于 7 天的日志文件
find /var/log -type f -name "*.log" -mtime +7 -exec ls -lh {} \;

踩坑总结

  1. dfdu 结果不一致,通常是有已删除未释放的文件
  2. lsof +L1 可以找出已删除但还被占用的文件
  3. 重启进程可以释放文件句柄
  4. 配置日志轮转(logrotate)可以预防此类问题
  5. 定期监控磁盘使用率,设置告警阈值

这个问题的坑在于:df 说满了,du 说没满。理解了 Linux 的文件删除机制,排查就很清晰了。

容器化部署的 Python 服务,跑在 K8s 上,平时没问题,流量一大就开始报 ConnectionPool is full。Pod 自动扩容后好了,缩容又报。这篇文章记录一下容器环境下的排查过程。

问题现象

服务部署在 K8s 上,HPA(Horizontal Pod Autoscaler)根据 CPU 使用率自动扩缩容。高峰期 Pod 数量从 3 扩到 10,这时候开始报错:

requests.exceptions.ConnectionError: ConnectionPool(pool_size=10, connections=10, maxsize=10, blocking=False)

坑在于:Pod 扩容后,每个 Pod 都有 20 个 worker 线程,但连接池只有 10 个位置。10 个 Pod 同时调用后端,连接池瞬间满了。

排查过程

第一步:查看 Pod 日志

# 查看 Pod 日志
kubectl logs -f deployment/backend-client --tail=100

# 看到大量 ConnectionPool is full 错误

第二步:确认后端状态

# 检查后端服务是否正常
kubectl exec -it deployment/backend-client -- curl -s http://backend-service:8080/health
# 返回 200,后端没问题

第三步:检查连接池配置

import requests

# 查看当前 Session 的连接池配置
s = requests.Session()
for host, adapter in s.adapters.items():
    print(f"Host: {host}, Pool: {adapter._pool_connections}, Max: {adapter._pool_maxsize}")

发现问题:每个 Pod 的连接池大小是 10,但每个 Pod 有 20 个 worker。10 个 Pod 同时调用,连接池瞬间满了。

问题原因

说白了,就是容器扩缩容导致并发数突增,连接池容量不够。

K8s HPA 扩容时,新 Pod 启动后立即开始处理请求。如果每个 Pod 有 20 个 worker,10 个 Pod 就是 200 个并发。但连接池只有 10 个位置,根本不够用。

坑在于:连接池大小是静态配置的,不会随着 Pod 数量动态调整。

解决方案

方案一:根据 Pod 数量动态调整连接池

import os
import requests
from requests.adapters import HTTPAdapter

# 从环境变量获取 Pod 数量(或 worker 数量)
pod_count = int(os.environ.get("POD_COUNT", 1))
workers_per_pod = int(os.environ.get("WORKERS_PER_POD", 20))

# 动态计算连接池大小
pool_size = pod_count * workers_per_pod

session = requests.Session()
adapter = HTTPAdapter(pool_connections=pool_size, pool_maxsize=pool_size)
session.mount("https://", adapter)
session.mount("http://", adapter)

方案二:使用共享连接池(推荐)

如果后端服务支持 HTTP/2,可以使用共享连接池:

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# 配置重试策略
retry = Retry(
    total=3,
    backoff_factor=0.1,
    status_forcelist=[500, 502, 503, 504],
)

adapter = HTTPAdapter(
    pool_connections=20,
    pool_maxsize=20,
    max_retries=retry,
)
session.mount("https://", adapter)
session.mount("http://", adapter)

方案三:限制并发数

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    # 限制并发数为 10
    semaphore = asyncio.Semaphore(10)

    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

asyncio.run(main())

生产环境最佳实践

  1. 连接池大小 >= 总并发数:总并发数 = Pod 数量 × 每个 Pod 的 worker 数
  2. 使用 HTTP/2:HTTP/2 支持多路复用,一个连接可以并发多个请求
  3. 监控连接池状态:在 Prometheus 里监控连接池使用率
  4. 设置合理的超时timeout=(3, 10) 连接超时 3 秒,读取超时 10 秒
# 生产环境推荐配置
import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 动态计算连接池大小
pod_count = int(os.environ.get("POD_COUNT", 1))
workers_per_pod = int(os.environ.get("WORKERS_PER_POD", 20))
pool_size = pod_count * workers_per_pod

session = requests.Session()

retry = Retry(total=3, backoff_factor=0.1, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(pool_connections=pool_size, pool_maxsize=pool_size, max_retries=retry)
session.mount("https://", adapter)
session.mount("http://", adapter)

踩坑总结

  1. 容器扩缩容会导致并发数突增,连接池容量不够
  2. 连接池大小是静态配置的,不会随 Pod 数量动态调整
  3. 总并发数 = Pod 数量 × 每个 Pod 的 worker 数
  4. 使用 HTTP/2 可以减少连接数
  5. 生产环境要监控连接池状态

这个问题的坑在于:本地测试没问题,上了 K8s 就报错。关键是理解容器环境下的并发模型,根据实际 Pod 数量和 worker 数量配置连接池。