Python技术迷

你还在使用打桩来记录for循环吗?

昨晚我在楼下便利店排队买咖啡,前面那哥们儿一直刷手机,队伍不动,我脑子就飘了……想到白天有人在群里问我:“东哥,我 for 循环里打了 20 个 print,怎么越跑越慢,还看不清到底卡哪了?”我当时差点脱口而出一句:你还在用打桩记录 for 循环啊……就是那个,怎么说呢,打印这玩意儿吧,它不是“免费”的。

你想啊,你在循环里 print(i),表面是记个点,实际是每次都在做 IO。IO 一慢,循环就跟着抖。更坑的是你一边跑一边打桩,日志刷屏,关键的那几条反而被淹了,最后你盯着终端发呆:我到底卡在 1273 还是 1274 来着……对吧。

我先给你看个很常见的“打桩写法”,别笑,我也干过,谁还没年轻过:

items = range(1_000_000)

for i, x in enumerate(items):
# 假装这里很复杂
    y = (x * 3) % 97

    print("i=", i, "x=", x, "y=", y)  # 经典打桩

你看着挺踏实,但它会把你 CPU 的节奏打乱。尤其你在服务器上跑,stdout 可能被收集到日志系统里,写磁盘、转发、聚合,链路一长,循环直接变“蜗牛爬”。而且多线程/异步环境下,打印还会交错,像菜市场一样,越看越烦。

我一般在工作里会换成三种更“像人”的办法,反正就是别让 print 绑架你。

第一种,最省事:抽样打印。就是别每次都打,隔一段打一次,而且别只打 i,要打“能定位问题”的东西,比如耗时、异常计数、最近一个样本值之类。

import time

items = range(1_000_000)
t0 = time.perf_counter()
last = t0

for i, x in enumerate(items, 1):
    y = (x * 3) % 97

# 每 50000 次报一次平安
if i % 50_000 == 0:
        now = time.perf_counter()
        total = now - t0
        chunk = now - last
        last = now
        speed = 50_000 / chunk if chunk > 0else float("inf")
        print(f"[beat] i={i} total={total:.2f}s chunk={chunk:.2f}s speed~{speed:.0f}/s last_y={y}")

这个就很“够用”了,至少不会刷屏,而且你能看到速度是不是突然掉了。掉了你再把粒度缩小,比如 5 万改 1 万,慢慢逼近。这个思路跟排查线上超时一样的,就是先定位“在哪一段慢”,再细化,不是上来就把屏幕打满。

第二种,比 print 更正经一点:logging + 等级控制。好处是你可以在开发时开 DEBUG,线上只留 INFO/WARN,不用改代码到处删 print。还有就是 logging 你能加时间、线程、文件行号啥的,定位起来舒服。

import logging
import time

logger = logging.getLogger("loop")
handler = logging.StreamHandler()
fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
handler.setFormatter(fmt)
logger.addHandler(handler)
logger.setLevel(logging.INFO)  # 想更啰嗦就改成 DEBUG

defevery(seconds: float):
"""简单限流器:控制日志多久打一次"""
    last = 0.0
defok():
nonlocal last
        now = time.perf_counter()
if now - last >= seconds:
            last = now
returnTrue
returnFalse
return ok

ok = every(0.5)  # 半秒最多打一条

items = range(1_000_000)
t0 = time.perf_counter()

for i, x in enumerate(items, 1):
    y = (x * 3) % 97

if ok():
        logger.info("i=%d elapsed=%.2fs y=%d", i, time.perf_counter() - t0, y)

这个“半秒打一条”特别适合那种循环里每次都很快的任务,你想观察进度,又不想输出爆炸。你甚至可以加个“异常计数”,比如遇到解析失败就 err += 1,然后日志里把 err 打出来,一眼就知道是不是数据质量问题在发作。

第三种,稍微硬核一点点:用计时器把循环拆成段。很多人打桩的真正诉求不是“看 i”,而是“我想知道到底慢在哪个步骤”。那你就把步骤包一层计时器,不用 print i 了,直接看各段耗时占比。

import time
from contextlib import contextmanager

@contextmanager
defspan(name: str, threshold: float = 0.0):
    t0 = time.perf_counter()
try:
yield
finally:
        cost = time.perf_counter() - t0
if cost >= threshold:
            print(f"[span] {name} cost={cost:.4f}s")

defstep_a(x):
return (x * 3) % 97

defstep_b(y):
# 假装这里是 IO 或者复杂计算
return (y ** 2) % 123457

items = range(200_000)

for i, x in enumerate(items, 1):
with span("step_a", threshold=0.002):
        y = step_a(x)

with span("step_b", threshold=0.002):
        z = step_b(y)

# 只有当某一步突然慢过阈值,才会吐出来
if i == 1:
pass

你看这个就很“干净”,平时不吵你,只有当某一步超过阈值才报警。很多线上偶发毛刺,就靠这种“阈值吐日志”抓出来的。因为偶发的慢,不会每次发生,你用 print(i) 根本抓不到“为什么慢”,只会抓到“慢的时候刚好 i=xxx”,那没啥用啊。

哦对了,还有个很现实的坑:你用 print 打桩,经常会改着改着忘了删,然后代码进了仓库,下一次上线,日志量爆炸,运维同学半夜喊你起来……你说尴尬不尴尬。logging 这种至少能用级别关掉,属于给自己留后路。

再啰嗦一句啊,有人会说“那我就 print(..., flush=True) 更及时”。嗯……及时是及时了,但 flush 会更慢,等于你每次都强制把缓冲刷出去,性能更惨。就像你排队买咖啡,你每走一步都要回头确认一下队伍有没有动,那你肯定更慢,对吧。