工控软件日志系统设计：分级、轮转、检索，别再 printf 了
引言
很多上位机项目的日志是这样的：Console.WriteLine("读到数据了")，跑起来一屏滚，出了事翻不到，不出事占满盘。现场断电、通信抖动、批次混料，事后能还原真相的只有日志——它不是开发期的调试输出，是产线的黑匣子。这篇讲一套通用打法：分级怎么定、格式怎么写、文件怎么转、坏了怎么查，全部用 Python 标准库落地，换 C# 思路一样。
一、分级：先定语义，再定级别
日志分级的价值不在级别名字，在于每一级绑定一个动作。工控场景我建议这么定界：
- DEBUG：原始报文、轮询细节、中间变量。默认关闭，排查通信问题时临时打开，量极大；
- INFO：业务事实——开工、完工、换型、批次切换、班次汇总。这是事后追溯的主食；
- WARNING：程序自己处理了的异常——通信超时后重连成功、单次写库失败后重试成功。记录但不打扰人；
- ERROR：需要人介入的事——重连 N 次仍失败、数据库打不开、设备状态与预期不一致。
最常见的错误是把所有异常都打成 ERROR。通信抖一下重连成功了，也 ERROR，一晚上几千条，真正要命的报警被淹死。判断标准就一句话：这条日志出来，需不需要有人看到。不需要就是 WARNING 及以下。
分级落地要支持全局阈值和模块阈值两层：全局 INFO，通信模块可以单独压到 WARNING（正式量产后轮询报文没信息量），调试时单独放开到 DEBUG。
二、格式：写结构化行，别写散文
2026-09-26 10:32 PLC连不上了怎么办啊——这种日志检索等于肉眼。结构化日志用 JSON 行（JSONL），每行一个对象，固定字段加业务字段：
importlogging,logging.handlers,json,queue,time,atexitclassJsonFormatter(logging.Formatter):"""每条日志一行 JSON：时间/级别/模块/消息 + 业务上下文(dev设备/ev事件)"""defformat(self,rec):returnjson.dumps({"ts":time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(rec.created)),"lvl":rec.levelname,"mod":rec.name,"msg":rec.getMessage(),"dev":getattr(rec,"dev",""),# 业务字段：设备编号"ev":getattr(rec,"ev",""),# 业务字段：事件码},ensure_ascii=False)defbuild_logger(path="logs/app.log"):q=queue.SimpleQueue()root=logging.getLogger()root.setLevel(logging.INFO)root.addHandler(logging.handlers.QueueHandler(q))# 业务线程只往队列丢，不碰磁盘fh=logging.handlers.RotatingFileHandler(path,maxBytes=10*1024*1024,backupCount=10,encoding="utf-8")fh.setFormatter(JsonFormatter())listener=logging.handlers.QueueListener(q,fh,respect_handler_level=True)listener.start()atexit.register(listener.stop)returnroot log=build_logger()log.info("批次开工",extra={"dev":"PLC-01","ev":"batch_start"})log.warning("通信超时，重连成功",extra={"dev":"PLC-01","ev":"poll_timeout"})两个设计点：一是extra带结构化业务字段（设备号、事件码），检索时能按字段过滤，不用正则猜；二是QueueHandler 把写盘挪到后台线程——采集主循环里同步写文件，磁盘一抖就拖慢轮询周期，工控机上这是真事故源。
三、轮转：日志写满盘 = 二次事故
用 RotatingFileHandler 按大小轮转（上面maxBytes+backupCount），10MB × 10 份封顶 100MB。三个量要按现场算：日志速率 × 保留天数 ≤ 磁盘余量的十分之一。想多留历史，压缩归档旧文件，别加份数。
工控机磁盘小，日志把盘写满的后果是数据库也写不进去——一次通信抖动放大成整线停机。所以轮转上限不是可选项，是系统设计的一部分。
四、检索：为"十分钟内出结论"设计
有了 JSONL，检索就是过滤。现场要的答案通常是三类：某设备最近半小时的 WARNING、某个时间段全厂 ERROR、某事件码的出现次数。一个 30 行脚本全覆盖：
importjson,datetimeasdtdefquery(path,level=None,dev=None,minutes=30,tail=50):cut=dt.datetime.now()-dt.timedelta(minutes=minutes)out=[]withopen(path,encoding="utf-8")asf:forlineinf:try:r=json.loads(line)exceptjson.JSONDecodeError:continue# 半行坏记录(断电截断)直接跳过ifdt.datetime.strptime(r["ts"],"%Y-%m-%d %H:%M:%S")<cut:continueiflevelandr["lvl"]!=level:continueifdevandr["dev"]!=dev:continueout.append(line.rstrip())returnout[-tail:]forlinquery("logs/app.log",level="WARNING",dev="PLC-01"):print(l)注意json.JSONDecodeError那行：断电最容易把最后一行日志截成半行，检索代码必须容忍，不然排查工具自己先崩。
五、三个避坑
- 时间戳用统一时钟。多台工控机时间不齐，日志对不上时间轴。开机对时，时间戳带毫秒更好；
- 日志里永远不写口令和密钥。配置里的连接串、授权码，打印前先抹掉——日志文件会被原样发出去排障，走到哪泄露到哪；
- 别在 ERROR 里只写"失败"两个字。上下文（设备号、参数、当时状态）一次带全，现场没有第二次复现机会。
要点回顾
- 分级绑定动作：ERROR=要人介入，自愈的异常降为 WARNING；
- JSONL 结构化格式 +
extra业务字段，检索按字段过滤不靠正则； - QueueHandler 异步写盘，轮转封顶，日志永远不能写满盘；
- 检索脚本容忍半行坏记录，为断电后十分钟出结论设计。
这套东西在我参与的采集、测试、报表类项目里是同一份代码复制着用的。系列前作写过配置管理和报警系统设计，日志体系是同一层基础设施——报警通知链的"事后可查"、远程排障的"有据可依"，都建立在日志先记对的基础上。
完整可运行的日志模块（Python 版含按模块动态调级 + 压缩归档，另附 C# NLog 等价配置）放在作者付费资源里，拿去就能挂进现有工程。
