你肯定遇到过这种时刻一台服务几小时前还正常突然端口起不来一眼扫过去 netstat 刷出一整屏连接可就是不知道这些都来自哪个进程。我在一次线上事故里翻了 20 多个进程才定位到罪魁祸首当时就决定用 Python 写一个把进程和网络连接串起来的监控与分析工具。它做的事很简单调起 psutil一次性采集系统里所有进程的元信息、所有 TCP/UDP 连接的端口与远端地址再用 PID 做纽带把它们缝合在一起最后提供端口反查、连接聚合、外联统计、差异监控这几类最常用视角。它能解决的痛点是传统系统工具的数据割裂——netstat 只能给你行数据任务管理器只能给你进程数据没人帮你回答“8080 被谁占了”“哪个进程连着几十个外网 IP”这类问题。这套工具适合运维排查故障、开发调试端口冲突也适合安全同学快速看机器上有没有异常外联就算你是 Python 新手跟着文章把每个 API 跑一遍对系统底层的理解也会上一截。它只做只读采集绝不干预进程调度也不修改任何连接状态可以放心在自己机器上反复试。1. 为什么有现成工具还要自己用 Python 写一个1.1 系统自带工具的三块短板先说结论系统自带工具不是不能用而是“够看不够用”。Windows 用户最常用的是 netstat -ano 和任务管理器。netstat -ano 输出了 PID但你要想知道这个 PID 对应的进程叫啥还得去任务管理器里找任务管理器默认只展示用户名、CPU、内存根本不展示 cmdline 和网络连接的关系。后来微软在资源监视器里补上了“网络”标签能按进程看 TCP 连接可它的刷新频率、筛选能力都很弱你没法把结果拉出来做二次处理。Linux 下有 ss、lsof功能很强大可它们同样是命令行工具输出格式是为了给人看不是为了给脚本 parse你想知道 10 分钟前哪个进程连了多少个不同的远端 IP单靠这几个命令很难得到答案。数据割裂是第一块短板。运维排障时通常会经历这样的链路先定位“哪个进程在大量占用连接”然后看这个进程的 CPU、内存、启动时间、状态再顺着进程去分析它连了哪些 IP、哪些端口。这个链路里至少需要 netstat、任务管理器、资源监视器三个工具来回切换同步 PID 全靠眼睛。如果进程频繁启停手工对 PID 的速度根本追不上变化。第二块短板是分析视角缺失。系统工具给你的是原始行数据而不是答案。我想要的是“连接数倒序 Top 10 的进程”“每个进程分别连了多少个不同的远端 IP”“两个时刻之间新增了哪些进程”这类报表没有原生工具支持。安全排查场景里要求按进程聚合外联一条一条连接根本看不过来得先算好再给你看。第三块短板是无法自动化与沉淀。线上环境需要定时采样、记录趋势、设置阈值告警这是一个脚本或服务才能干的事。用 Python 把采集逻辑沉淀成函数想接告警接告警想落库落库想加 Web 界面加 Web 界面这些能力系统工具一个都不具备。1.2 技术选型psutil 为什么是首选为什么选 psutil因为它的定位就是“跨平台系统信息采集”进程、网络、磁盘、传感器全都覆盖。你也可以直接调系统 API但 Linux 下读 /proc/net/tcp 时IP 是以反序十六进制存放的端口也是十六进制地址映射、用户 ID 解析都得自己造轮子Windows 下更麻烦要调用 NtQuerySystemInformation 这类内部接口。psutil 把这些差异全部封装好我只需要面对统一的 Python 对象。之前我在一个采集脚本里试过直接解析 ss 的文本输出光处理段边界和列宽就耗费了一个下午最后还挂在 IPv6 地址的冒号与方括号上。从性能上说psutil 不是最快的直接读 /proc 肯定更快但它足够可靠。一次全量 net_connections 在普通机器上也就几十毫秒级别做监控完全够用。如果你追求极致性能应该去写 eBPF而不是用 Python。做这类工具开发效率、可维护性、跨平台能力通常比那几毫秒更重要。方案跨平台编程接口二次分析性能适用场景psutilWindows / Linux / macOS好Python 原生对象方便中需要写脚本、定时分析、自定义报表netstat / ss一般无需解析文本一般高临时瞄一眼连接状态lsofLinux / macOS 为主无弱中单机端口反查任务管理器资源监视器Windows无弱低人肉观察单个进程的交互界面从生态成熟度看psutil 是 Python 系统监控的事实标准很多开源监控 Agent 的数据层都用它。文档极其稳定踩坑资料也多长期维护成本低。我在多个项目里见过它的身影说明这条路是被验证过的。2. 数据模型如何把“进程”和“网络连接”缝合起来2.1 先吃透两个核心 API动手写代码之前先把两个核心 API 讲清楚。第一个是 psutil.process_iter(attrs)。它遍历系统里的进程和 process.list 相比process_iter 是流式的进程特别多时内存压力更小。attrs 参数很有用它指定要预加载的字段这些字段会存进 proc.info 字典里之后访问 proc.info[name] 不会再触发系统调用。如果你不传 attrs而是对每个进程逐个调 proc.name()、proc.cmdline()每个方法都是一次独立系统调用几百个进程顺着调一遍速度差异会让你崩溃。我一般至少带上 pid、name、username、cmdline、create_time 这五样后面两个在分析进程变化时尤其有用。第二个是 psutil.net_connections(kindinet)。它返回系统里当前所有网络连接kind 参数控制范围inet 表示 TCP 加 UDP 的 IPv4/IPv6排查网络基本都用它tcp 只取 TCPunix 可以拿本机进程间通信的 unix socket。返回的每条记录是 namedtuple字段有 fd、family、type、laddr、raddr、status、pid。这里有个细节很容易踩如果调用 psutil.Process.connections()返回记录里没有 pid 字段必须结合 Process 对象本身去理解而全局的 net_connections() 反而带 pid这给我们用 PID 做关联提供了最大便利。family 对应 socket.AF_INET / AF_INET6type 对应 SOCK_STREAM / SOCK_DGRAMladdr 和 raddr 为空元组时表示这条连接还没有绑定本地地址或远端地址。状态字段在 TCP 下常见的是 LISTEN、ESTABLISHED、TIME_WAIT、CLOSE_WAIT、SYN_SENTUDP 没有连接概念status 一般是空字符串。如果你对“Electron 主进程与渲染进程之间怎么通过 IPC 通信”这类问题感兴趣把 kind 换成 unix就能看到本机内部 socket 的路径和 fd很多软件的多进程协作、守护进程与业务进程的本地通信都会在这里留下痕迹这也是排查进程间关系时一个冷门但有效的视角。2.2 关联逻辑PID 就是数据库主键全量连接好拿难的是把它和进程信息拼起来。我的做法是把 pid 当作数据库主键。先建一个 pid 到 Process 信息的字典再用 net_connections 返回的 pid 去查字典查得到就补上 name/cmdline/username/create_time查不到就把该条记录标记为 unknown。查不到的原因通常是两个进程在采样间隙退出了或者这条连接由内核发起、pid 字段为 None。这里有一个关键的性能取舍先拿全量连接再通过 pid 反查进程字典而不是反过来先遍历每个进程、再逐个取 connections。后一种写法在进程数几百、连接数几万时会非常糟糕等于把连接列表重复扫描了很多遍。反过来一次 net_connections 拿到全部连接进程信息用字典缓存整体复杂度接近 O(连接数 进程数)。cmdline 建议做截断。真实环境里一个 Java 进程的 cmdline 可能几百个字符全打出来会把其他信息淹没。我在实现里只保留前 120 个字符足够识别进程身份又不会刷屏。create_time 保留为浮点时间戳需要的时候再用当前时间减去它换算成运行时长。进程数量大时用户名 username 的收集通常会触发额外系统查询所以我只在实在需要时才把它加入 attrs。2.3 权限边界与异常容忍策略系统监控绕不开权限这个话题。在 Linux 下没有 root 权限时/proc 里其他用户进程的 fd 目录不可读psutil 访问这些进程的连接或 cmdline 时会抛 AccessDeniedWindows 下 System 进程、部分系统服务同样需要管理员权限才能看到细节。所以这个工具要么以管理员/root 运行要么在代码里把所有采集异常都捕获掉允许部分进程跳过。实际排查时我经常先不加权限跑一遍输出里提示“有 N 个进程因权限被跳过”就够了大部分业务场景已经有结论。另一个高频异常是 NoSuchProcess进程在两次 API 调用之间退出这是常态而不是 bug。处理原则只有一条——单进程采集失败绝不中断全量采集。网络连接列表里如果出现拿不到进程信息的 pid把它保留为“未知进程”记录同样不算失败。这个设计哲学贯穿整个工具监控程序是旁观者得先保证自己不死才能谈得上描述系统。3. 完整实现从采集层到分析层再到交互层3.1 采集层一次遍历拿到进程与连接的全量视图先给出采集层代码。这一步的目标很纯粹返回一个干净的快照结构包含所有连接、连接的进程信息、按进程聚合的连接数以及采样过程中被跳过多少条。代码可以直接跑唯一的外部依赖是 psutil。import socket import time from collections import defaultdict import psutil def collect_snapshot(): 采集一次快照进程信息 IP 网络连接信息。 proc_map {} for proc in psutil.process_iter([pid, name, username, cmdline, create_time]): try: # .info 是 process_iter 预加载好的字段字典 proc_map[proc.pid] proc.info except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue sessions [] skipped 0 try: raw_conns psutil.net_connections(kindinet) except psutil.AccessDenied: print([warn] net_connections() 权限不足请改用管理员/root 运行) raw_conns [] for conn in raw_conns: pinfo proc_map.get(conn.pid) sessions.append({ pid: conn.pid, proc_name: (pinfo or {}).get(name, ), username: (pinfo or {}).get(username, ), cmdline: .join((pinfo or {}).get(cmdline, []) or [])[:120], create_time: (pinfo or {}).get(create_time, None), family: IPv4 if conn.family socket.AF_INET else IPv6, type: TCP if conn.type socket.SOCK_STREAM else UDP, laddr_ip: conn.laddr.ip if conn.laddr else , laddr_port: conn.laddr.port if conn.laddr else 0, raddr_ip: conn.raddr.ip if conn.raddr else , raddr_port: conn.raddr.port if conn.raddr else 0, status: conn.status, }) if pinfo is None: skipped 1 per_proc defaultdict(int) for s in sessions: per_proc[(s[pid], s[proc_name])] 1 return { sessions: sessions, per_proc: per_proc, skipped: skipped, proc_count: len(proc_map), proc_pids: set(proc_map.keys()), }proc_map 的键是 pid值是 process_iter 预装载好的 info 字典。这里用 pinfo.get() 而不是直接取是因为连接对应进程可能已经退出pinfo 可能是 None。raddr 为空时代表无远端比如 LISTEN 状态的监听 socket 就没有远端地址。UDP 的 status 通常为空展示时按空字符串处理即可。3.2 分析层端口反查、连接聚合、外联统计分析层每个函数都以 snapshot 为唯一输入。这样采集层和分析层解耦后续加新的统计口径不会改到底层代码。先说端口反查排查“8080 被谁占了”时要匹配 laddr_port同时也要看 raddr_port因为一条连接只要一端是目标端口就跟你查的端口相关。排序用状态字段LISTEN 排最前这符合直觉。def lookup_port(snapshot, port): rows [s for s in snapshot[sessions] if s[laddr_port] port or s[raddr_port] port] rows.sort(keylambda x: (x[status], x[pid])) return rows def top_conn_processes(snapshot, top10): 返回连接数最多的进程列表。 items sorted(snapshot[per_proc].items(), keylambda x: x[1], reverseTrue) return items[:top] def outbound_summary(snapshot): 按进程统计外联远端 IP 数量识别可疑外联。 agg defaultdict(set) for s in snapshot[sessions]: if s[type] TCP and s[status] ESTABLISHED and s[raddr_ip]: agg[(s[pid], s[proc_name])].add(s[raddr_ip]) result [] for (pid, name), ips in agg.items(): result.append({ pid: pid, name: name, remote_ip_count: len(ips), remote_ips: sorted(ips), }) result.sort(keylambda x: x[remote_ip_count], reverseTrue) return result[:50]外联统计这块我要重点说它统计的是“每个进程连了多少个不同的远端 IP”而不是“多少条连接”。两者含义完全不同。一个爬虫进程可能开了几千条连接到同一个目标 IP按连接数看很夸张但其实很规律真正异常的是连接数不太大、但远端 IP 数量一下子涨起来的进程。安全排查时我一般先跑 outbound_summaryremote_ip_count 异常高的立刻拿去对进程名和 cmdline 验证。top_conn_processes 则用于快速找出连接数最多的服务。3.3 交互层四类子命令撑起日常运维写命令行入口时我用 argparse 的子命令来组织snapshot 打印一次快照port 按端口反查top 查看连接数排行watch 持续监控进程变化。代码不做花哨的表格渲染直接用字符串对齐。watch 模式的间隔参数受一个下限保护至少 1 秒防止有人把 interval 设成 0.1 把机器卡住。def print_rows(rows, limit20): for s in rows[:limit]: print( f{s[pid]:7} {s[proc_name]:20} {s[type]:4} f{s[status]:10} {s[laddr_ip]}:{s[laddr_port]} - f{s[raddr_ip]}:{s[raddr_port]} ) def main(): import argparse parser argparse.ArgumentParser(descriptionPython 进程网络监控与分析工具) sub parser.add_subparsers(destcommand) p1 sub.add_parser(snapshot, help打印一次全量快照) p1.add_argument(--show, choices[all, tcp, udp, established, listen], defaultall) p2 sub.add_parser(port, help按端口反查进程例如 port 8080) p2.add_argument(port, typeint) p3 sub.add_parser(top, help查看连接数最多的进程) p3.add_argument(-n, typeint, default10) p4 sub.add_parser(watch, help持续监控进程变化) p4.add_argument(-i, --interval, typefloat, default2.0) p4.add_argument(-n, typeint, default10) args parser.parse_args() snap collect_snapshot() if args.command snapshot: rows snap[sessions] if args.show tcp: rows [s for s in rows if s[type] TCP] elif args.show udp: rows [s for s in rows if s[type] UDP] elif args.show established: rows [s for s in rows if s[status] ESTABLISHED] elif args.show listen: rows [s for s in rows if s[status] LISTEN] print_rows(rows, limit30) print(f\ntotal connections: {len(rows)}, skipped: {snap[skipped]}, processes: {snap[proc_count]}) elif args.command port: rows lookup_port(snap, args.port) print_rows(rows, limit30) if rows else print(fport {args.port} 当前没有匹配的连接) elif args.command top: for (pid, name), count in top_conn_processes(snap, args.n): print(f{pid:7} {name:24} {count}) elif args.command watch: prev set(snap[proc_pids]) while True: cur collect_snapshot() new_pids cur[proc_pids] - prev gone_pids prev - cur[proc_pids] if new_pids: print(f[] 新增进程: {sorted(new_pids)[:10]}) if gone_pids: print(f[-] 退出进程: {sorted(gone_pids)[:10]}) print(--- Top 连接数进程 ---) for (pid, name), count in top_conn_processes(cur, args.n): print(f{pid:7} {name:24} {count}) prev cur[proc_pids] time.sleep(max(args.interval, 1.0)) if __name__ __main__: main()这个入口我实际跑下来很顺手。比如查 8080 端口python netprobe.py port 8080持续盯进程python netprobe.py watch -i 2。watch 模式的核心价值在对比两个时刻的进程集合服务启动失败、子进程异常退出、出现全新进程都会在几秒内被标记出来。如果你在 Windows 上看到一些同名进程同时出现比如微信或者 Chrome 的多进程不要惊讶那是软件自身多进程设计和 IPC 协作的结果工具里用 create_time 和 cmdline 就能区分谁先谁后。3.4 兼容性细节与运行建议跨平台兼容不是一句空话。第一件事是 fd 字段Windows 上 net_connections 返回的 fd 恒为 -1展示时直接忽略不要试图拿它做任何判断。第二件事是 IPv6 监听地址Linux 下监听所有 IPv6 地址时 laddr.ip 显示为 ::展示时可以映射为 *避免和空值混淆。第三件事是 UDP 的 status 字段没有 ESTABLISHED所以外联统计要显式限定 type TCP否则 UDP 的“状态”会被误判。运行环境方面Python 3.6 和 psutil5.8 就够了。安装不复杂pip install psutil 一条命令。如果你命令行环境还没配好 Python先装一个干净的 Python 3把 pip 路径加到 PATH 里再安装在虚拟环境里也好、全局也好都行。工具本身没有 GUI输出是纯文本直接放到服务器上也能跑。权限建议我再说一遍本机单用户场景一般能拿到大部分进程要全量连接和所有进程的 cmdlineLinux 下用 sudoWindows 下用管理员身份打开终端。工具不会自动提权也不建议自动提权提示一句让操作者自己决定比偷偷申请管理员权限更安全。4. 踩坑实录那些文档不会告诉你的排查细节4.1 高频异常逐个拆解第一个坑是 NoSuchProcess。这个异常不只是发生在遍历中process_iter 预装载字段后到访问 proc.info[name] 时进程可能已经没了。所以在采集循环里要整体 try。短命进程非常常见比如 curl 请求一次、启动脚本里的临时命令它们往往在一两秒内就结束你的工具刚扫到它就已经退出了。遇到这种情况跳过就好别让一条记录毁了整个快照。第二个坑是 AccessDenied。Windows 上 System 进程、某些杀毒软件的保护进程即使在管理员控制台下也会有访问限制Linux 上其他用户的进程在非 root 下同样受保护。这类进程通常也会有自己的网络连接看不到确实遗憾但工具要做的不是死磕而是把权限信息打到输出里跳过几条、为什么跳过、建议怎么补全。我看到不少开源工具因为这里没处理好在权限不足的环境里直接崩溃非常可惜。第三个坑是 ZombieProcess。僵尸进程已经没有资源了它的 fd、网络连接全是空访问它 parent 关系以外的字段很容易抛异常。捕获并 skip 是标准做法。还有一个隐藏坑是命令行编码Windows 上进程的 cmdline 可能含有非 UTF-8 字节直接 join 可能抛 UnicodeDecodeError。稳妥的做法是在拼接字符串时对每个元素做 encoding 纠错处理或者在 print 输出时兜底。我常用的一段兜底函数长这样def safe_cmdline(proc): try: return .join(proc.cmdline()) except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess, UnicodeDecodeError): return 4.2 端口占用排查的几个细节端口起不来的报错基本都是“Address already in use”这时候第一个要确认的是这个端口处于什么状态。如果是一个进程正常 LISTEN那就是真正的端口冲突去进程列表找罪魁祸首如果是大量 TIME_WAIT那是历史连接在 NAT 或 TIME_WAIT 周期里残留Linux 默认要等 60 秒左右重启业务进程也无济于事。直接用工具查端口时把 status 列看清楚再下结论。Windows 下有些端口是被 System 进程PID 4占的这类通常是系统 HTTP.sys 或 Hyper-V 的网络栈服务不能简单杀进程只能改配置。Linux 下也有类似情况比如内核线程监听在某些端口上。工具只负责告诉你“哪个 PID 占了它”接下来要不要动、怎么动需要结合业务知识。还有个小细节如果你的服务监听的是 0.0.0.0:8080但外部访问却走不通这时候端口反查能看到进程在监听问题大概率出在防火墙或云安全组而不在进程本身。工具能帮你排除进程这一层剩下的排查范围瞬间缩小。4.3 性能与资源占用如何控制psutil 的 net_connections 是一次全量遍历普通笔记本上十几毫秒可服务器连接数达到几十万时单次开销会放大到几百毫秒甚至更久。因此我强烈建议监控间隔不要低于 1 秒watch 模式的默认 2 秒在生产环境是合理的。连续高频采样除了增加 CPU 开销还会放大 /proc 的 IO 压力得不偿失。进程信息的收集同样有成本。cmdline、username 这类属性每个都要读一次 /proc进程多时很可观。用 process_iter 传 attrs 预装载是第一步优化更进一步的优化是区分“快速模式”和“详细模式”快速模式只取 name、cmdline 两个字段详细模式才查 username 和 create_time。当前快照工具面向排查场景一次采样拿全量足够用。还有一点容易被忽略watch 模式里面不要每次都重新构建全量快照来做对比。正确做法是缓存上一次的 proc_pids 或连接列表本次只增量对比。上面的代码就是这么实现的prev 集合保存在外面循环里 collect_snapshot再比较差异。这种写法虽然简单但比“每秒全量 dump 然后重算”省得多。4.4 问题速查表现象常见原因处理建议提示 net_connections() 权限不足未以管理员/root 身份运行Linux 用 sudoWindows 用管理员终端确认后再跑部分进程的 cmdline 为空Linux 内核线程没有 cmdlineWindows 权限不足或进程已退出用进程名兜底允许跳过统计 skipped 数量port 反查没结果但端口确实被占用可能是 TIME_WAIT 残留或进程权限不足读不到用 --show listen 只看监听状态提升权限再跑IPv6 监听地址显示 ::表示绑定所有 IPv6 地址展示时把 :: 映射为 *也可以和 IPv4 0.0.0.0 合并显示进程列表出现大量同名进程业务本身采用多进程设计比如浏览器、即时通讯、Java 网关结合 create_time 和 cmdline 区分守护进程与工作进程连接数巨大导致采样卡顿系统连接数十几万net_connections 单次遍历开销大增大采样间隔按需把 kind 改为 tcp4 等更小范围这张表是我根据真实使用反馈整理出来的两条最常遇到权限不足和 TIME_WAIT 误解。权限好办TIME_WAIT 那个往往让新手以为端口冲突其实等一两分钟或者用 sysctl 调整参数就能解决工具只是帮你把状态看清楚。5. 后续还能怎么扩展这个工具目前是命令行交互但真正的监控系统一定不止于此。第一个扩展方向是告警在快照里发现某进程连接数突增、出现陌生进程监听公网端口时通过企业微信或钉钉 webhook 推一条消息到群里。实现不复杂无非是 collect_snapshot 后加一组规则检查再调用 webhook 客户端。第二个方向是数据落库每秒一次快照存到 sqlite按小时跑聚合报表可以看到连接数趋势、远程 IP 变化轨迹给安全分析或者容量规划都很有价值。第三个方向是 Web 界面。基础数据仍然是 collect_snapshot前端用一个轻量 Python 服务定期拉快照再用表格和图表呈现。这里要说一句不要先做界面再做数据层顺序反了会导致界面重做。我见过太多项目上来先搞仪表盘结果采集逻辑一堆洞最后还是得回来补基础。第四个方向是把它做成常驻服务Linux 下配 systemd unitWindows 下做成计划任务或服务采集逻辑保持只读长时期稳定运行。结合进程守护和进程间通信IPC模式这个工具还能把快照结果转给其他程序做联动处理比如收到告警后自动收集系统日志、导出 netstat 摘要。我个人在实际操作里的体会是监控工具难的不是界面、告警或者花哨功能而是把进程、连接、状态变化这几个维度干净地建模把所有权限异常稳稳兜住。我后来把这套脚本改了 N 版日常排查端口冲突、确认某个进程是不是连着不该连的外网 IP、甚至看一台机器上有没有无人认领的孤儿进程都是先跑一遍快照再下结论。如果你也照着写我建议先别急着加图表和告警把 collect_snapshot 和异常处理写稳后面所有功能都是在同一份干净数据上做聚合。手里有个趁手的小工具排查问题就不至于海底捞针了。
