简介这是一套面向网络管理员、系统运维人员与数据分析师的Python网络流量监控可视化系统。基于Python 3.7开发能够实时采集与展示网络传输数据支持流量、数据包、延迟、连接数等多维度分析并内置基于统计方法的异常检测与协议分析功能可用于日常网络巡检、性能评估与故障排查场景对网络运维场景尤为实用。压缩包共9个文件包含4个Python源码主程序、数据生成器、可视化模块及演示脚本、3张PNG效果图实时流量看板、性能热力图、网络分析图以及依赖清单与中文说明文档整体大小仅1.63MB目录结构清晰便于直接运行、调试与二次开发。资源已有114人学习下载内置的模拟数据生成器可快速体验完整流程从实时监控到异常告警均有示例适合Python可视化与网络数据分析学习者参考实践。1. 网络流量监控可视化的核心思路实时图表比事后分析更依赖数据管道大多数团队接到流量监控需求时第一反应是上 Prometheus 加 Grafana或者直接买商业探针。但如果你只是想把一台服务器的出入口流量、每秒包数和 TCP 连接状态变成一张自己说了算的实时数据图表用 Python 从零搭一个轻量展示系统完全可行而且比想象中简单。这个场景的本质是 Python 数据分析与可视化的实时化延伸核心难点并不在绘图库选哪个而在采样、差分、缓存、刷新这四个环节的衔接顺序。顺序错了曲线要么是锯齿状要么延迟半分钟以上。下面从采集层讲到可视化层给出可以直接跑的 psutil 秒级采样代码、deque 环形缓冲数据结构以及基于 pyecharts 的页面刷新方案最后补齐多网卡过滤、窗口容量和验证方法这些生产环境才会遇到的细节。2. 采集层psutil 网卡计数与抓包方案的选型2.1 psutil 的数据来源/proc/net/dev 与差分采样原理Linux 上做网络流量监控第一条路是读内核维护的网卡累计计数器。/proc/net/dev每个网卡一行内容是接收字节数、数据包数、错误数、丢包数以及发送方向的同一套字段。psutil 的net_io_counters(pernicTrue)就是对这组计数器的跨平台封装Windows 和 macOS 上走各自系统 API对外接口保持一致这也是它在 python 生态里被选作监控基座的主要原因。这里的关键认知是内核给的是从开机到现在的累计值不是速率。速率必须靠两次采样做差分再除以间隔时间。间隔取 1 秒得到的数值是每秒字节数换算成 Mbps 要乘 8 再除 1e6。很多人第一次写出的监控曲线锯齿严重原因往往不在采集频率而是差分前没判断计数器是否发生跳变。网卡重置、虚拟机热插拔、驱动 reload 都会让累计值瞬间变小不做保护就会算出负速率或者一个虚假的极大峰值。另一个容易忽略的点是回环接口 lo。默认不带pernic参数时返回所有网卡合计本机进程互访的流量会被重复计入导致上行和下行同时暴涨对故障判断产生严重干扰。做可视化大屏时一般只挑物理网卡展示或者至少把 lo 单独画一条线而不是混进总吞吐里。2.2 可直接运行的秒级差分采样代码下面是采集层最小可运行版本psutil 安装好就能跑不依赖任何框架import time import psutil def sample_net(interval1.0): prev psutil.net_io_counters(pernicTrue) time.sleep(interval) curr psutil.net_io_counters(pernicTrue) rows {} for nic, c in curr.items(): p prev.get(nic) if p is None: continue # 计数器回绕或网卡重置时丢弃本次差分防止出现负速率 if c.bytes_recv p.bytes_recv or c.bytes_sent p.bytes_sent: continue rows[nic] { rx_mbps: (c.bytes_recv - p.bytes_recv) * 8 / interval / 1e6, tx_mbps: (c.bytes_sent - p.bytes_sent) * 8 / interval / 1e6, rx_pps: c.packets_recv - p.packets_recv, tx_pps: c.packets_sent - p.packets_sent, } return rows if __name__ __main__: for _ in range(5): print(sample_net(1.0)) time.sleep(1)采样逻辑说明函数先取一次快照睡满一个完整间隔再取第二次快照差分值除以间隔就是该秒平均速率。* 8 / 1e6把字节换算成兆比特网络设备厂商和机房报表习惯用 Mbps图表坐标轴统一用它避免和 MB/s 混淆。packets_recv的差分直接就是每秒包数抓突增或广播风暴时包数曲线往往比字节曲线更早暴露异常。注意这个函数的 sleep 会阻塞调用线程正式接入数据管道时采样逻辑应该独立一个线程运行展示层不能和它抢同一个循环。提示pernicTrue必须带上否则拿不到单网卡维度也就无法做后续的多网卡过滤和业务名映射。首次调用返回的快照是基准点第二次调用开始才有有效差分值。2.3 报文级采集sniff 与 AF_PACKET 的取舍计数器方案拿不到哪个远端 IP 在打流量这类信息这时需要报文级采集。scapy 的sniff(prncallback, storeFalse)是最常见的原型写法但纯 Python 对每个包做协议解析开销很大万兆网卡满速抓包时丢包是必然的。真实系统里更常见的做法是 tcpdump 落盘 pcap 再做离线分析或者用 dpkt 配合原始套接字自己拆二层头import socket import dpkt sock socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(0x0003)) while True: data, addr sock.recvfrom(65535) eth dpkt.ethernet.Ethernet(data) if not isinstance(eth.data, dpkt.ip.IP): continue ip eth.data print(addr[0], ip.src, ip.dst, len(data))这段代码只做二层解析拿到入口网卡、源 IP、目的 IP 和帧长比 scapy 轻一个量级。0x0003表示接收所有以太网协议类型recvfrom的缓冲区要留足默认 8KB 在巨型帧下会截断数据包。前提是具备 root 权限普通用户打不开 AF_PACKET 套接字。wireshark 抓下来的 pcap 文件也能用 dpkt 逐包解析喂进同一套可视化管线只是实时性换成了离线分析。选型上可以按这个表对号入座维度psutil 计数器AF_PACKET 原始套接字scapy sniff采集粒度网卡级累计数据链路层帧解析后的报文对象CPU 开销极低中高可观察信息吞吐、包数、丢包自行拆出五元组协议解析最方便典型场景长时间实时曲线轻量流量审计原型验证、离线分析2.4 多网卡监控时的字段映射与虚拟网卡问题docker 环境或 KVM 宿主机里net_io_counters会列出 docker0、veth 开头的一堆虚拟网卡这些接口的流量其实已经在物理网卡里计入过一次全部加总会把数值虚高一倍以上。我一般用一个排除规则只保留以 eth、ens、enp 开头的物理网卡并把内核网卡名映射成业务标签图表的图例直接显示公网出口内网这类名字而不是让值班人员去猜 ens192 是什么。import psutil import re PHYSICAL re.compile(r^(eth|enp|ens|eno)\d*) nic_labels {eth0: 公网出口, eth1: 内网} def visible_nics(): counters psutil.net_io_counters(pernicTrue) return {nic_labels.get(n, n): v for n, v in counters.items() if PHYSICAL.match(n)}正则说明^(eth|enp|ens|eno)覆盖传统 eth0 命名和 systemd 的可预测命名规范\d*适配带数字后缀的变体。物理网卡命名在不同发行版差异很大先把psutil.net_io_counters(pernicTrue).keys()打印出来看一眼再配正则不要盲抄规则。标签映射放字典里维护新增机器时不需要改采集代码。3. 数据管道环形缓冲与时间窗口聚合3.1 为什么用 deque 做环形缓冲而不是 List采集层每秒产生一条记录可视化层每 2 到 3 秒取一次最近数据。如果直接把所有点都塞进 List运行几小时后内存里全是历史点而图表根本不需要那么长的窗口。环形缓冲用固定长度覆盖最旧数据Python 的collections.deque(maxlenN)就是这个语义尾部追加和头部淘汰都是 O(1)不需要手动pop(0)引起整体搬移。操作deque(maxlenN)普通 List尾部追加O(1)O(1)头部淘汰O(1)自动覆盖pop(0) 为 O(n)内存上限固定 N 个元素无限增长并发快照需拷贝成 List需拷贝成 List窗口长度的设定和刷新频率直接耦合。每秒采样、展示最近 300 个点曲线覆盖 5 分钟想看小时级趋势要么把 maxlen 调大导致绘图变慢要么做降采样聚合。所以缓冲长度本质上是时间跨度和绘制性能之间的折中我一般会把原始秒级数据保留 600 点另外维护一份分钟级聚合列表用于长时间观察。3.2 基于 dataclass 的采样点与缓冲代码用 dataclass 定义采样点比裸 dict 更省心字段即文档后面接可视化层时按属性取值也不会打错 keyfrom collections import deque from dataclasses import dataclass import time dataclass class TrafficPoint: ts: float # 采样时间戳epoch 秒 rx: float # 下行 Mbps tx: float # 上行 Mbps class RingBuffer: def __init__(self, maxlen600): self._buf deque(maxlenmaxlen) def append(self, p: TrafficPoint): self._buf.append(p) def tail(self, n120): return list(self._buf)[-n:] def __len__(self): return len(self._buf)缓冲逻辑说明deque(maxlen600)在插入超过容量的元素时自动丢弃最左侧的旧点所以 append 不需要额外判断长度。tail(n)返回最近 n 个点的浅拷贝原因是 deque 在绘制线程遍历期间如果被采样线程追加元素可能拿到不一致的快照拷贝成 List 后绘图层拿到的是一个静止的视图不会再报deque mutated during iteration这类错误。生产环境建议给 append 和 tail 加同一个 threading.Lock或者直接用 queue.Queue 解耦。采样线程只负责 put可视化线程只负责 get这样即使图表渲染卡了 500ms采样节奏也不会被打乱。3.3 长时段查看时的降采样聚合窗口拉长以后每秒一个点的曲线在 1440 像素宽的屏幕上本来也画不出 86400 个点的差异反而拖慢浏览器渲染。分钟级聚合是把时间戳整除窗口宽度组内求和取平均属于典型的时序降采样from collections import defaultdict def downsample(points, window60): buckets defaultdict(lambda: [0.0, 0.0, 0]) for p in points: key int(p.ts // window) * window buckets[key][0] p.rx buckets[key][1] p.tx buckets[key][2] 1 return [{ ts: key, rx_avg: v[0] / v[2], tx_avg: v[1] / v[2], } for key, v in sorted(buckets.items())]聚合逻辑说明ts // window把同一分钟内的所有点归到同一个桶* window还原成该分钟起点的时间戳方便图表 x 轴对齐。窗口越大曲线越平滑但峰值会被平均掉。监控场景里这恰恰是需要注意的持续 10 秒的突增流量放在 5 分钟平均曲线里可能完全看不出异常所以聚合表不能只存平均值要把max一起算出来或者保证异常告警走原始窗口而不是聚合窗口。3.4 多进程场景Redis List 与单进程直连的边界展示系统如果要把多台采集节点的数据汇总到一台机器展示单进程 deque 就不够用了。常见做法是采集进程把 JSON 行推到 Redis List 左侧展示层用lrange取尾部数据。Redis 侧要注意三点List 要做长度裁剪防止无限增长采样端用 pipeline 批量写入减少往返 RTT时间戳由采集端统一生成不要在展示端补否则跨机器的时钟偏移会让多条曲线在同一 x 坐标上错位。单机单网卡、秒级采样、三小时以内窗口deque 完全够用引入消息队列反而增加故障面。这是选型时最容易犯的过度设计先确认规模再上组件而不是先搭一套 Kafka 再往里灌每秒一条的数据。4. 可视化层实时图表的刷新机制与图表选型4.1 实时刷新三种做法的对比可视化层的选择其实卡在刷新机制上不在图表库本身。任何前端图表都能刷新真正的差异是数据怎么从 Python 到浏览器。三种常见做法按复杂度递增排列第一种是 matplotlib FuncAnimation完全本地绘制适合桌面脚本和临时验证缺点是不适合做成团队可访问的页面。第二种是 pyecharts 服务端重渲染 HTML浏览器整页定时刷新代码量最小秒级刷新足够用缺点是整页重载有白屏闪烁。第三种是 plotly Dash 的dcc.Interval组件回调里只更新图表数据不重载页面体验最好但依赖体积和工程复杂度也最大。方案刷新机制适合场景依赖规模matplotlib FuncAnimation画布内增量绘制本地脚本、内网小屏小pyecharts 页面定时刷新服务端重渲染 HTML轻量展示系统中plotly Dash dcc.Interval前端回调拉数据重交互大屏大4.2 matplotlib FuncAnimation 秒级动态曲线FuncAnimation 的思路是每隔固定毫秒调用一次更新函数函数内部清空坐标轴再重绘最新窗口。样本数据和更新函数通过闭包关联缓冲区由采集线程填充import matplotlib.pyplot as plt from matplotlib.animation import FuncAnimation import psutil import time from collections import deque buf_rx deque(maxlen120) buf_tx deque(maxlen120) tick deque(maxlen120) t0 time.time() def poll(): prev psutil.net_io_counters() time.sleep(1) curr psutil.net_io_counters() buf_rx.append((curr.bytes_recv - prev.bytes_recv) * 8 / 1e6) buf_tx.append((curr.bytes_sent - prev.bytes_sent) * 8 / 1e6) tick.append(time.time() - t0) def update(frame): poll() ax.clear() ax.plot(list(tick), list(buf_rx), label下行, color#d14) ax.plot(list(tick), list(buf_tx), label上行, color#1a7) ax.legend(locupper right) ax.set_xlabel(相对秒数) ax.set_ylabel(Mbps) ax.set_title(f网络流量实时监控 下行 {buf_rx[-1]:.2f} Mbps) fig, ax plt.subplots(figsize(10, 4)) ani FuncAnimation(fig, update, interval1000, cache_frame_dataFalse) plt.show()刷新机制说明interval1000表示每 1000ms 触发一次 updateupdate 里先 poll 推进一个采样点再重绘图表的节奏由动画定时器驱动而不是 sleep 驱动窗口拖拽时不会卡死采样节奏。cache_frame_dataFalse是关键参数默认 True 会缓存每一帧的数据长时间运行内存持续上涨加上它才能按帧丢弃。时间轴用相对秒数而不是绝对时间戳避免 x 轴标签过长挤在一起。4.3 pyecharts 服务端渲染与页面定时刷新的展示系统要做成浏览器能访问的展示系统pyecharts 加一个轻量 HTTP 服务是最省事的路径。重交互大屏那套在这里用不上能看、能刷新、能长期挂着就足够from datetime import datetime from flask import Flask, Response from pyecharts.charts import Line from pyecharts import options as opts app Flask(__name__) buf RingBuffer(600) # 复用第 3 章的环形缓冲 app.route(/traffic) def traffic_chart(): points buf.tail(120) line Line() line.add_xaxis([datetime.fromtimestamp(p.ts).strftime(%H:%M:%S) for p in points]) line.add_yaxis(下行, [round(p.rx, 2) for p in points], is_smoothTrue) line.add_yaxis(上行, [round(p.tx, 2) for p in points], is_smoothTrue) line.set_global_opts( title_optsopts.TitleOpts(title网络流量实时监控), yaxis_optsopts.AxisOpts(nameMbps), datazoom_optsopts.DataZoomOpts(range_start0, range_end100), ) return Response(line.render_embed(), mimetypetext/html) app.route(/) def index(): return (meta http-equivrefresh content3 iframe src/traffic stylewidth:100%;height:95vh;border:0/iframe) if __name__ __main__: app.run(host0.0.0.0, port8080, threadedTrue)页面刷新机制说明meta http-equivrefresh content3让浏览器每 3 秒整页重载一次每次重载重新请求/traffic服务端从环形缓冲取出最近 120 个点渲染成新图表。如果采样是 1 秒一次页面 3 秒刷新那么每次刷新推进 3 个点曲线是跳跃式前进而不是平滑滚动改成 1 秒刷新体验更好但服务端渲染压力直接高三倍。threadedTrue必须开启否则页面请求阻塞时采样进程也拿不到锁。注意render_embed()会把 echarts 的 JS 库内联进 HTML每次刷新整段 JS 都会重新下载。内网环境问题不大公网低带宽环境下建议改用手动引入 echarts.min.js 的方式页面体积能从几百 KB 降到几十 KB。4.4 刷新间隔、数据窗口与防抖参数实时图表有三个参数需要配套调整采样间隔、页面刷新间隔、展示窗口长度。推荐起点是 1 秒采样、3 秒刷新、120 点窗口这套参数下曲线完整覆盖 6 分钟每次刷新只新增 3 个点浏览器渲染压力很小。要扩大到 1 小时窗口把窗口加到 3600 个点后前端折线图每帧渲染已经能感觉到卡顿这时应该回到第 3 章的降采样聚合而不是无脑加长原始窗口。防抖在这个场景里不是指鼠标点击而是多次刷新触发同一次重绘。pyecharts 的重渲染是全量生成 HTML没有增量机制连续请求会被浏览器排队处理。如果监控页面越跑越慢先看是不是有人开了多个标签页同时轮询其次把 refresh 的 content 值从 3 调整到 5牺牲一点实时性换取稳定刷新。5. 验证与排错多网卡、性能边界和常见坑5.1 用 /proc/net/dev 手工核对采样值接入真实网络前先做一致性验证。开一个终端跑watch -n 1 cat /proc/net/dev | grep eth0另一个终端跑自己的采样脚本对比两次输出的字节增量是否一致。重点核对差分方向字节计数是累计值采样脚本必须每次记录上一次快照任何一次漏记都会让瞬时速率虚高一倍。如果采样值是内核计数字段值的两倍基本可以确定是把累计值当成了增量直接绘图。5.2 多网卡监控的过滤与重命名第 2 章提过 docker0 和 veth 会重复计数过滤规则不要写死在代码里拆到配置里维护。物理网卡命名在不同发行版差异很大先打印psutil.net_io_counters(pernicTrue).keys()再配正则。虚拟网卡和回环接口全部过滤后剩下的一般只有两三个物理网卡这时用网卡名到业务标签的映射字典把图例从 ens192 变成公网出口监控页面的可读性会明显上一个台阶。5.3 采样频率与缓冲容量怎么设采样频率不是越高越好。1 秒采样已经能覆盖绝大多数异常识别场景0.5 秒采样的价值只是突增检测提前半秒但代价是数据量翻倍、曲线抖动更明显。建议 1 秒起步确有必要再加密。缓冲容量按公式容量 展示秒数 / 采样间隔计算600 容量对应 10 分钟窗口加大容量之前先确认可视化层能扛住这个点数。提示一条实用经验是原始细粒度数据保留 10 分钟用于实时排障聚合数据保留 24 小时用于趋势观察。两套窗口分别设置容量不要试图用一个超大 deque 覆盖所有场景。5.4 时间戳对齐与可视化大屏适配的两个细节多节点汇总后把各节点时间戳对齐到同一秒再入缓冲否则折线图同一个 x 值上会画出多条错位的线。对齐在采集端完成取int(time.time())丢弃毫秒即可展示端不要做任何补时操作。可视化大屏适配通常做两件事图表宽度用百分比而不是固定像素这样render_embed输出的 HTML 才能在 1080p 和 4K 屏上自适应y 轴范围固定而不是自动缩放否则流量一波动坐标轴就来回跳值班人员的视觉疲劳会显著上升。固定范围设成理论带宽的 80%超过上限本身就是一种告警呈现方式。本文还有配套的精品资源点击获取
Python实时网络流量监控：从psutil采样到pyecharts可视化
简介这是一套面向网络管理员、系统运维人员与数据分析师的Python网络流量监控可视化系统。基于Python 3.7开发能够实时采集与展示网络传输数据支持流量、数据包、延迟、连接数等多维度分析并内置基于统计方法的异常检测与协议分析功能可用于日常网络巡检、性能评估与故障排查场景对网络运维场景尤为实用。压缩包共9个文件包含4个Python源码主程序、数据生成器、可视化模块及演示脚本、3张PNG效果图实时流量看板、性能热力图、网络分析图以及依赖清单与中文说明文档整体大小仅1.63MB目录结构清晰便于直接运行、调试与二次开发。资源已有114人学习下载内置的模拟数据生成器可快速体验完整流程从实时监控到异常告警均有示例适合Python可视化与网络数据分析学习者参考实践。1. 网络流量监控可视化的核心思路实时图表比事后分析更依赖数据管道大多数团队接到流量监控需求时第一反应是上 Prometheus 加 Grafana或者直接买商业探针。但如果你只是想把一台服务器的出入口流量、每秒包数和 TCP 连接状态变成一张自己说了算的实时数据图表用 Python 从零搭一个轻量展示系统完全可行而且比想象中简单。这个场景的本质是 Python 数据分析与可视化的实时化延伸核心难点并不在绘图库选哪个而在采样、差分、缓存、刷新这四个环节的衔接顺序。顺序错了曲线要么是锯齿状要么延迟半分钟以上。下面从采集层讲到可视化层给出可以直接跑的 psutil 秒级采样代码、deque 环形缓冲数据结构以及基于 pyecharts 的页面刷新方案最后补齐多网卡过滤、窗口容量和验证方法这些生产环境才会遇到的细节。2. 采集层psutil 网卡计数与抓包方案的选型2.1 psutil 的数据来源/proc/net/dev 与差分采样原理Linux 上做网络流量监控第一条路是读内核维护的网卡累计计数器。/proc/net/dev每个网卡一行内容是接收字节数、数据包数、错误数、丢包数以及发送方向的同一套字段。psutil 的net_io_counters(pernicTrue)就是对这组计数器的跨平台封装Windows 和 macOS 上走各自系统 API对外接口保持一致这也是它在 python 生态里被选作监控基座的主要原因。这里的关键认知是内核给的是从开机到现在的累计值不是速率。速率必须靠两次采样做差分再除以间隔时间。间隔取 1 秒得到的数值是每秒字节数换算成 Mbps 要乘 8 再除 1e6。很多人第一次写出的监控曲线锯齿严重原因往往不在采集频率而是差分前没判断计数器是否发生跳变。网卡重置、虚拟机热插拔、驱动 reload 都会让累计值瞬间变小不做保护就会算出负速率或者一个虚假的极大峰值。另一个容易忽略的点是回环接口 lo。默认不带pernic参数时返回所有网卡合计本机进程互访的流量会被重复计入导致上行和下行同时暴涨对故障判断产生严重干扰。做可视化大屏时一般只挑物理网卡展示或者至少把 lo 单独画一条线而不是混进总吞吐里。2.2 可直接运行的秒级差分采样代码下面是采集层最小可运行版本psutil 安装好就能跑不依赖任何框架import time import psutil def sample_net(interval1.0): prev psutil.net_io_counters(pernicTrue) time.sleep(interval) curr psutil.net_io_counters(pernicTrue) rows {} for nic, c in curr.items(): p prev.get(nic) if p is None: continue # 计数器回绕或网卡重置时丢弃本次差分防止出现负速率 if c.bytes_recv p.bytes_recv or c.bytes_sent p.bytes_sent: continue rows[nic] { rx_mbps: (c.bytes_recv - p.bytes_recv) * 8 / interval / 1e6, tx_mbps: (c.bytes_sent - p.bytes_sent) * 8 / interval / 1e6, rx_pps: c.packets_recv - p.packets_recv, tx_pps: c.packets_sent - p.packets_sent, } return rows if __name__ __main__: for _ in range(5): print(sample_net(1.0)) time.sleep(1)采样逻辑说明函数先取一次快照睡满一个完整间隔再取第二次快照差分值除以间隔就是该秒平均速率。* 8 / 1e6把字节换算成兆比特网络设备厂商和机房报表习惯用 Mbps图表坐标轴统一用它避免和 MB/s 混淆。packets_recv的差分直接就是每秒包数抓突增或广播风暴时包数曲线往往比字节曲线更早暴露异常。注意这个函数的 sleep 会阻塞调用线程正式接入数据管道时采样逻辑应该独立一个线程运行展示层不能和它抢同一个循环。提示pernicTrue必须带上否则拿不到单网卡维度也就无法做后续的多网卡过滤和业务名映射。首次调用返回的快照是基准点第二次调用开始才有有效差分值。2.3 报文级采集sniff 与 AF_PACKET 的取舍计数器方案拿不到哪个远端 IP 在打流量这类信息这时需要报文级采集。scapy 的sniff(prncallback, storeFalse)是最常见的原型写法但纯 Python 对每个包做协议解析开销很大万兆网卡满速抓包时丢包是必然的。真实系统里更常见的做法是 tcpdump 落盘 pcap 再做离线分析或者用 dpkt 配合原始套接字自己拆二层头import socket import dpkt sock socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.ntohs(0x0003)) while True: data, addr sock.recvfrom(65535) eth dpkt.ethernet.Ethernet(data) if not isinstance(eth.data, dpkt.ip.IP): continue ip eth.data print(addr[0], ip.src, ip.dst, len(data))这段代码只做二层解析拿到入口网卡、源 IP、目的 IP 和帧长比 scapy 轻一个量级。0x0003表示接收所有以太网协议类型recvfrom的缓冲区要留足默认 8KB 在巨型帧下会截断数据包。前提是具备 root 权限普通用户打不开 AF_PACKET 套接字。wireshark 抓下来的 pcap 文件也能用 dpkt 逐包解析喂进同一套可视化管线只是实时性换成了离线分析。选型上可以按这个表对号入座维度psutil 计数器AF_PACKET 原始套接字scapy sniff采集粒度网卡级累计数据链路层帧解析后的报文对象CPU 开销极低中高可观察信息吞吐、包数、丢包自行拆出五元组协议解析最方便典型场景长时间实时曲线轻量流量审计原型验证、离线分析2.4 多网卡监控时的字段映射与虚拟网卡问题docker 环境或 KVM 宿主机里net_io_counters会列出 docker0、veth 开头的一堆虚拟网卡这些接口的流量其实已经在物理网卡里计入过一次全部加总会把数值虚高一倍以上。我一般用一个排除规则只保留以 eth、ens、enp 开头的物理网卡并把内核网卡名映射成业务标签图表的图例直接显示公网出口内网这类名字而不是让值班人员去猜 ens192 是什么。import psutil import re PHYSICAL re.compile(r^(eth|enp|ens|eno)\d*) nic_labels {eth0: 公网出口, eth1: 内网} def visible_nics(): counters psutil.net_io_counters(pernicTrue) return {nic_labels.get(n, n): v for n, v in counters.items() if PHYSICAL.match(n)}正则说明^(eth|enp|ens|eno)覆盖传统 eth0 命名和 systemd 的可预测命名规范\d*适配带数字后缀的变体。物理网卡命名在不同发行版差异很大先把psutil.net_io_counters(pernicTrue).keys()打印出来看一眼再配正则不要盲抄规则。标签映射放字典里维护新增机器时不需要改采集代码。3. 数据管道环形缓冲与时间窗口聚合3.1 为什么用 deque 做环形缓冲而不是 List采集层每秒产生一条记录可视化层每 2 到 3 秒取一次最近数据。如果直接把所有点都塞进 List运行几小时后内存里全是历史点而图表根本不需要那么长的窗口。环形缓冲用固定长度覆盖最旧数据Python 的collections.deque(maxlenN)就是这个语义尾部追加和头部淘汰都是 O(1)不需要手动pop(0)引起整体搬移。操作deque(maxlenN)普通 List尾部追加O(1)O(1)头部淘汰O(1)自动覆盖pop(0) 为 O(n)内存上限固定 N 个元素无限增长并发快照需拷贝成 List需拷贝成 List窗口长度的设定和刷新频率直接耦合。每秒采样、展示最近 300 个点曲线覆盖 5 分钟想看小时级趋势要么把 maxlen 调大导致绘图变慢要么做降采样聚合。所以缓冲长度本质上是时间跨度和绘制性能之间的折中我一般会把原始秒级数据保留 600 点另外维护一份分钟级聚合列表用于长时间观察。3.2 基于 dataclass 的采样点与缓冲代码用 dataclass 定义采样点比裸 dict 更省心字段即文档后面接可视化层时按属性取值也不会打错 keyfrom collections import deque from dataclasses import dataclass import time dataclass class TrafficPoint: ts: float # 采样时间戳epoch 秒 rx: float # 下行 Mbps tx: float # 上行 Mbps class RingBuffer: def __init__(self, maxlen600): self._buf deque(maxlenmaxlen) def append(self, p: TrafficPoint): self._buf.append(p) def tail(self, n120): return list(self._buf)[-n:] def __len__(self): return len(self._buf)缓冲逻辑说明deque(maxlen600)在插入超过容量的元素时自动丢弃最左侧的旧点所以 append 不需要额外判断长度。tail(n)返回最近 n 个点的浅拷贝原因是 deque 在绘制线程遍历期间如果被采样线程追加元素可能拿到不一致的快照拷贝成 List 后绘图层拿到的是一个静止的视图不会再报deque mutated during iteration这类错误。生产环境建议给 append 和 tail 加同一个 threading.Lock或者直接用 queue.Queue 解耦。采样线程只负责 put可视化线程只负责 get这样即使图表渲染卡了 500ms采样节奏也不会被打乱。3.3 长时段查看时的降采样聚合窗口拉长以后每秒一个点的曲线在 1440 像素宽的屏幕上本来也画不出 86400 个点的差异反而拖慢浏览器渲染。分钟级聚合是把时间戳整除窗口宽度组内求和取平均属于典型的时序降采样from collections import defaultdict def downsample(points, window60): buckets defaultdict(lambda: [0.0, 0.0, 0]) for p in points: key int(p.ts // window) * window buckets[key][0] p.rx buckets[key][1] p.tx buckets[key][2] 1 return [{ ts: key, rx_avg: v[0] / v[2], tx_avg: v[1] / v[2], } for key, v in sorted(buckets.items())]聚合逻辑说明ts // window把同一分钟内的所有点归到同一个桶* window还原成该分钟起点的时间戳方便图表 x 轴对齐。窗口越大曲线越平滑但峰值会被平均掉。监控场景里这恰恰是需要注意的持续 10 秒的突增流量放在 5 分钟平均曲线里可能完全看不出异常所以聚合表不能只存平均值要把max一起算出来或者保证异常告警走原始窗口而不是聚合窗口。3.4 多进程场景Redis List 与单进程直连的边界展示系统如果要把多台采集节点的数据汇总到一台机器展示单进程 deque 就不够用了。常见做法是采集进程把 JSON 行推到 Redis List 左侧展示层用lrange取尾部数据。Redis 侧要注意三点List 要做长度裁剪防止无限增长采样端用 pipeline 批量写入减少往返 RTT时间戳由采集端统一生成不要在展示端补否则跨机器的时钟偏移会让多条曲线在同一 x 坐标上错位。单机单网卡、秒级采样、三小时以内窗口deque 完全够用引入消息队列反而增加故障面。这是选型时最容易犯的过度设计先确认规模再上组件而不是先搭一套 Kafka 再往里灌每秒一条的数据。4. 可视化层实时图表的刷新机制与图表选型4.1 实时刷新三种做法的对比可视化层的选择其实卡在刷新机制上不在图表库本身。任何前端图表都能刷新真正的差异是数据怎么从 Python 到浏览器。三种常见做法按复杂度递增排列第一种是 matplotlib FuncAnimation完全本地绘制适合桌面脚本和临时验证缺点是不适合做成团队可访问的页面。第二种是 pyecharts 服务端重渲染 HTML浏览器整页定时刷新代码量最小秒级刷新足够用缺点是整页重载有白屏闪烁。第三种是 plotly Dash 的dcc.Interval组件回调里只更新图表数据不重载页面体验最好但依赖体积和工程复杂度也最大。方案刷新机制适合场景依赖规模matplotlib FuncAnimation画布内增量绘制本地脚本、内网小屏小pyecharts 页面定时刷新服务端重渲染 HTML轻量展示系统中plotly Dash dcc.Interval前端回调拉数据重交互大屏大4.2 matplotlib FuncAnimation 秒级动态曲线FuncAnimation 的思路是每隔固定毫秒调用一次更新函数函数内部清空坐标轴再重绘最新窗口。样本数据和更新函数通过闭包关联缓冲区由采集线程填充import matplotlib.pyplot as plt from matplotlib.animation import FuncAnimation import psutil import time from collections import deque buf_rx deque(maxlen120) buf_tx deque(maxlen120) tick deque(maxlen120) t0 time.time() def poll(): prev psutil.net_io_counters() time.sleep(1) curr psutil.net_io_counters() buf_rx.append((curr.bytes_recv - prev.bytes_recv) * 8 / 1e6) buf_tx.append((curr.bytes_sent - prev.bytes_sent) * 8 / 1e6) tick.append(time.time() - t0) def update(frame): poll() ax.clear() ax.plot(list(tick), list(buf_rx), label下行, color#d14) ax.plot(list(tick), list(buf_tx), label上行, color#1a7) ax.legend(locupper right) ax.set_xlabel(相对秒数) ax.set_ylabel(Mbps) ax.set_title(f网络流量实时监控 下行 {buf_rx[-1]:.2f} Mbps) fig, ax plt.subplots(figsize(10, 4)) ani FuncAnimation(fig, update, interval1000, cache_frame_dataFalse) plt.show()刷新机制说明interval1000表示每 1000ms 触发一次 updateupdate 里先 poll 推进一个采样点再重绘图表的节奏由动画定时器驱动而不是 sleep 驱动窗口拖拽时不会卡死采样节奏。cache_frame_dataFalse是关键参数默认 True 会缓存每一帧的数据长时间运行内存持续上涨加上它才能按帧丢弃。时间轴用相对秒数而不是绝对时间戳避免 x 轴标签过长挤在一起。4.3 pyecharts 服务端渲染与页面定时刷新的展示系统要做成浏览器能访问的展示系统pyecharts 加一个轻量 HTTP 服务是最省事的路径。重交互大屏那套在这里用不上能看、能刷新、能长期挂着就足够from datetime import datetime from flask import Flask, Response from pyecharts.charts import Line from pyecharts import options as opts app Flask(__name__) buf RingBuffer(600) # 复用第 3 章的环形缓冲 app.route(/traffic) def traffic_chart(): points buf.tail(120) line Line() line.add_xaxis([datetime.fromtimestamp(p.ts).strftime(%H:%M:%S) for p in points]) line.add_yaxis(下行, [round(p.rx, 2) for p in points], is_smoothTrue) line.add_yaxis(上行, [round(p.tx, 2) for p in points], is_smoothTrue) line.set_global_opts( title_optsopts.TitleOpts(title网络流量实时监控), yaxis_optsopts.AxisOpts(nameMbps), datazoom_optsopts.DataZoomOpts(range_start0, range_end100), ) return Response(line.render_embed(), mimetypetext/html) app.route(/) def index(): return (meta http-equivrefresh content3 iframe src/traffic stylewidth:100%;height:95vh;border:0/iframe) if __name__ __main__: app.run(host0.0.0.0, port8080, threadedTrue)页面刷新机制说明meta http-equivrefresh content3让浏览器每 3 秒整页重载一次每次重载重新请求/traffic服务端从环形缓冲取出最近 120 个点渲染成新图表。如果采样是 1 秒一次页面 3 秒刷新那么每次刷新推进 3 个点曲线是跳跃式前进而不是平滑滚动改成 1 秒刷新体验更好但服务端渲染压力直接高三倍。threadedTrue必须开启否则页面请求阻塞时采样进程也拿不到锁。注意render_embed()会把 echarts 的 JS 库内联进 HTML每次刷新整段 JS 都会重新下载。内网环境问题不大公网低带宽环境下建议改用手动引入 echarts.min.js 的方式页面体积能从几百 KB 降到几十 KB。4.4 刷新间隔、数据窗口与防抖参数实时图表有三个参数需要配套调整采样间隔、页面刷新间隔、展示窗口长度。推荐起点是 1 秒采样、3 秒刷新、120 点窗口这套参数下曲线完整覆盖 6 分钟每次刷新只新增 3 个点浏览器渲染压力很小。要扩大到 1 小时窗口把窗口加到 3600 个点后前端折线图每帧渲染已经能感觉到卡顿这时应该回到第 3 章的降采样聚合而不是无脑加长原始窗口。防抖在这个场景里不是指鼠标点击而是多次刷新触发同一次重绘。pyecharts 的重渲染是全量生成 HTML没有增量机制连续请求会被浏览器排队处理。如果监控页面越跑越慢先看是不是有人开了多个标签页同时轮询其次把 refresh 的 content 值从 3 调整到 5牺牲一点实时性换取稳定刷新。5. 验证与排错多网卡、性能边界和常见坑5.1 用 /proc/net/dev 手工核对采样值接入真实网络前先做一致性验证。开一个终端跑watch -n 1 cat /proc/net/dev | grep eth0另一个终端跑自己的采样脚本对比两次输出的字节增量是否一致。重点核对差分方向字节计数是累计值采样脚本必须每次记录上一次快照任何一次漏记都会让瞬时速率虚高一倍。如果采样值是内核计数字段值的两倍基本可以确定是把累计值当成了增量直接绘图。5.2 多网卡监控的过滤与重命名第 2 章提过 docker0 和 veth 会重复计数过滤规则不要写死在代码里拆到配置里维护。物理网卡命名在不同发行版差异很大先打印psutil.net_io_counters(pernicTrue).keys()再配正则。虚拟网卡和回环接口全部过滤后剩下的一般只有两三个物理网卡这时用网卡名到业务标签的映射字典把图例从 ens192 变成公网出口监控页面的可读性会明显上一个台阶。5.3 采样频率与缓冲容量怎么设采样频率不是越高越好。1 秒采样已经能覆盖绝大多数异常识别场景0.5 秒采样的价值只是突增检测提前半秒但代价是数据量翻倍、曲线抖动更明显。建议 1 秒起步确有必要再加密。缓冲容量按公式容量 展示秒数 / 采样间隔计算600 容量对应 10 分钟窗口加大容量之前先确认可视化层能扛住这个点数。提示一条实用经验是原始细粒度数据保留 10 分钟用于实时排障聚合数据保留 24 小时用于趋势观察。两套窗口分别设置容量不要试图用一个超大 deque 覆盖所有场景。5.4 时间戳对齐与可视化大屏适配的两个细节多节点汇总后把各节点时间戳对齐到同一秒再入缓冲否则折线图同一个 x 值上会画出多条错位的线。对齐在采集端完成取int(time.time())丢弃毫秒即可展示端不要做任何补时操作。可视化大屏适配通常做两件事图表宽度用百分比而不是固定像素这样render_embed输出的 HTML 才能在 1080p 和 4K 屏上自适应y 轴范围固定而不是自动缩放否则流量一波动坐标轴就来回跳值班人员的视觉疲劳会显著上升。固定范围设成理论带宽的 80%超过上限本身就是一种告警呈现方式。本文还有配套的精品资源点击获取
