你还在使用打桩来记录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 会更慢,等于你每次都强制把缓冲刷出去,性能更惨。就像你排队买咖啡,你每走一步都要回头确认一下队伍有没有动,那你肯定更慢,对吧。