可运行的网络流量分析器源码：抓包、协议解析与流会话聚合

简介：面向网络运维与安全监控场景的网络流量分析器源码，基于VC++开发，适合具备C++基础、关注抓包流量统计实现的学习者与开发人员。程序可完成单点流量、点到点流量和协议流量排名，支持按流量统计与占比统计，并展示基于协议及目标端口的累计流量，同时提示ARP流量占比并给出ARP流量分析，可用于业务流量监控、病毒木马后门流量监控、ARP病毒监控和网络异常监控等场景，涉及内存表扫描、BarChart图表控件等关键技术点。资源包共36个文件，以7个h头文件、5个cpp源文件为核心，配套ico与bmp图标位图、rc资源脚本、sln/dsp/vcxproj等工程文件，以及mdb、xlsx、ini等数据与配置文件，压缩包约168KB，工程结构较完整，便于直接编译研究。目前已有909人学习下载，可作为流量分析、协议统计与图表展示模块的参考实现，帮助读者梳理抓包数据的统计口径、界面呈现方式及工程组织思路。

1. 网络流量分析器源码：一套能立刻跑起来的抓包与协议分析底座

前阵子排查一个接口偶发超时的问题，服务端日志一切正常，最后是靠抓包才看到是 TCP 重传在捣鬼。当时用的工具要么是图形界面、没法塞进自己的监控链路，要么就得从头写一套收包解析。这份网络流量分析器源码解决的正是这个尴尬：它把网卡收包、协议逐层解码、流会话重组、指标聚合和前端图表串成了一条完整链路，编译运行之后就能看到实时流量曲线和前 N 名会话排行。适合三类人：想给自家服务补一块流量面板的后端开发、做安全审计入门练习的同学、以及需要一个可改可调的抓包底座来做课程设计的开发者。它不是抓包工具的替代品，而是一个能被你改造成自己业务形状的骨架。

2. 技术栈与目录结构：从网卡收包到前端图表的整条链路

在动手跑之前，先把这条链路想清楚：数据包从网卡进入内核，被过滤器筛一遍，交给用户态程序解码，解码结果进流表做会话聚合，聚合结果再通过接口推给浏览器。每一段的技术选型都直接决定了后面能不能扛住真实流量，所以这一章先讲清楚为什么这么选，再落到目录结构和数据模型上。

2.1 选型逻辑：收包、解包、展示三段各用什么

收包这一段，Python 生态里有三条路。第一条是裸 socket，用AF_PACKET直接读网卡，性能最好但需要自己处理链路层头部和平台差异；第二条是绑定 libpcap，交给内核的 BPF 过滤器先做一轮筛选，用户态只处理活下来的包；第三条是 Scapy 的sniff，上手最快，但它是纯 Python 逐字段解析，线速场景下很容易成为瓶颈。这个项目的常见做法是第二条为主、第三条为辅：实时链路走 pcap 绑定加 BPF 过滤，离线调试和协议字段验证用 Scapy，两者共用同一套解码结果的数据结构，切换成本很低。

解包这一段是整套东西的核心。我的血泪经验是，别一上来就用重量级解析库处理每一个包。真实网络里大部分流量你根本不关心，先用 BPF 表达式在入口砍掉，比如只留ip and (tcp or udp)，剩下的再进分层解码器。解码器按「以太网头 → IP 头 → 传输层头 → 应用层特征」逐层剥，每层只读固定偏移的字段，读完就往下传，不做多余的对象构造。这样单核跑到几十万包每秒是能站住的，而同样流量交给通用解析库，CPU 很容易打满。

展示这一段，后端选的是轻量 HTTP 服务加 WebSocket。HTTP 负责配置管理、历史查询这类请求响应型操作，WebSocket 负责把每秒聚合出来的指标增量推给前端，避免前端高频轮询。前端用图表库画时序曲线和排行条形图，页面本身很薄，重点在数据。存储层用一份嵌入式数据库存会话摘要和分钟级指标快照，原始包默认不落盘，需要复盘时再开环形 pcap 文件写入。

提示：选型不是越新越好。抓包这类 IO 密集加 CPU 密集混合的场景，先把过滤做在入口，比换任何解析库都管用。

2.2 目录结构：每个包在链路里站哪一班岗

源码拆得比较干净，按职责分包，不按技术分层。下面这个结构是我整理后的样子，你拿到手基本就是这个形状：

traffic-analyzer/ ├── capture/ # 收包层 │ ├── engine.py # 抓包引擎，网卡选择、BPF、启停控制 │ ├── pcap_reader.py # 离线 pcap 文件读取，接口与引擎对齐 │ └── ring_writer.py # 环形 pcap 落盘，用于事后复盘 ├── decode/ # 解码层 │ ├── ethernet.py # 链路层 │ ├── ipv4.py # 网络层，分片处理在这里 │ ├── transport.py # TCP/UDP，端口与标志位 │ └── app_probe.py # 应用层特征识别（HTTP/DNS/TLS 握手） ├── flow/ # 会话层 │ ├── table.py # 五元组哈希表 │ └── session.py # 单条会话的状态与统计字段 ├── metrics/ # 聚合层 │ ├── window.py # 滑动窗口计数器 │ └── snapshot.py # 指标快照与 TopN 计算 ├── api/ # 接口层 │ ├── routes.py # REST 路由 │ └── ws.py # WebSocket 增量推送 ├── web/ # 前端静态资源 │ ├── index.html │ └── assets/ ├── config/ │ └── analyzer.yaml # 网卡、过滤器、阈值、保留时长 └── main.py # 入口，装配各层并启动

这个划分的好处是每一层都能单独测。capture层对外只暴露「给我一个包的回调」，decode层只依赖字节串不依赖网卡，flow和metrics是纯内存计算，可以喂构造好的包做单元测试，完全不需要 root 权限。很多人写这类工具喜欢把解析逻辑塞进抓包回调里，一行行往下写，最后想加个协议就得动主流程，这是最容易把自己坑住的结构。

2.3 数据模型：包、会话、指标快照三层对象

三层对象各自的字段要提前定好，后面所有模块都围着它们转。最底层是包记录，只保留分析必需的信息，绝不保存原始字节，否则内存撑不住；中间层是会话，用五元组做键，正反向合并成一条；最上层是指标快照，按固定时间窗口聚合出来，直接对应前端的一张图。

app_hint这个字段值得单独说一句。它不是在解码阶段硬判定协议，而是打一个「疑似 HTTP」「疑似 TLS 握手」的标记，交给上层的启发式规则去确认。这样做的好处是解码层保持薄，识别逻辑可以单独迭代，识别错了也不影响抓包主链路。

3. 核心模块拆解：抓包、协议解析、流重组与统计聚合

四个模块串起来就是一条数据流水线，任何一个环节慢都会拖垮整体。下面按数据流动的顺序逐个拆，每个模块给出关键代码和参数说明，你可以照着改。需要强调的是，这些代码都是从「能跑」出发的最小实现，生产环境还要加上错误处理和指标埋点。

3.1 抓包引擎：网卡选择、混杂模式与 BPF 过滤

抓包引擎要解决三件事：选哪块网卡、只收哪些包、怎么优雅地停。下面这段是启动逻辑，用 Scapy 的异步嗅探器做外壳，因为它帮我们处理了线程和停止信号，真正的过滤压力交给了内核 BPF：

from scapy.all import AsyncSniffer, conf class CaptureEngine: def __init__(self, iface: str, bpf_filter: str = "ip", on_packet=None): self.iface = iface # BPF 表达式在核心态执行，能砍掉大量无关流量，是最便宜的性能优化 self.bpf_filter = bpf_filter self.on_packet = on_packet self.sniffer = None def start(self): self.sniffer = AsyncSniffer( iface=self.iface, filter=self.bpf_filter, # 例如 "tcp port 80 or udp port 53" prn=self._handle, # 每个包的回调，运行在嗅探线程里 store=False, # 关键：不缓存原始包，否则内存直接爆 promisc=True, # 混杂模式，能收到非本机目的地的包 ) self.sniffer.start() def _handle(self, pkt): # 回调里只做最轻量的事：转交给解码队列，不做解析 if self.on_packet: self.on_packet(bytes(pkt)) def stop(self): if self.sniffer: self.sniffer.stop()

store=False是这里最容易被忽略的参数。默认情况下嗅探器会把收到的包全部存进列表，跑几分钟内存就上去了，很多人第一次跑就是栽在这个默认值上。promisc=True决定了你能不能看到交换机镜像口或者同网段其他机器的流量，如果只想看本机，关掉它更安全。prn回调里千万不要做解析，把它当成一个生产者，往队列里丢字节串就够了，解析放到独立消费者线程里做，否则抓包线程一卡，内核缓冲区溢出，你就会看到丢包计数蹭蹭涨。

网卡选择上，Linux 下用ip link看名字，无线网卡通常要额外设置监听模式才能看到全部帧，这个不在源码范围内。BPF 表达式建议先用tcpdump -i eth0 -d "你的表达式"验证一遍，确认编译通过再写进配置，否则启动时只会给你一句语焉不详的报错。

3.2 协议解析：以太网头到应用层的逐层解码

解码器的设计原则是「逐层剥、不回头」。每个函数只负责自己那一层，把结果填进同一个字典往下传，遇到不认识的协议就标记为 unknown 然后返回，绝不让异常冒到主循环：

import struct def decode_ethernet(data: bytes): if len(data) < 14: return None dst, src, eth_type = data[0:6], data[6:12], struct.unpack("!H", data[12:14])[0] # 0x0800 是 IPv4，0x86DD 是 IPv6，0x8100 是 VLAN 标签需要再剥 4 字节 if eth_type == 0x8100 and len(data) >= 18: eth_type = struct.unpack("!H", data[16:18])[0] return decode_ipv4(data[18:], eth_type) return decode_ipv4(data[14:], eth_type) def decode_ipv4(data: bytes, eth_type: int): if eth_type != 0x0800 or len(data) < 20: return {"proto": "unknown", "len": len(data)} ver_ihl = data[0] ihl = (ver_ihl & 0x0F) * 4 # 首部长度以 4 字节为单位 total_len = struct.unpack("!H", data[2:4])[0] proto = data[9] # 6=TCP, 17=UDP, 1=ICMP src = ".".join(str(b) for b in data[12:16]) dst = ".".join(str(b) for b in data[16:20]) frag_off = struct.unpack("!H", data[6:8])[0] & 0x1FFF if frag_off: # 分片包先标记，交给上层重组 return {"proto": "fragment", "src": src, "dst": dst, "len": total_len} return decode_transport(data[ihl:], proto, src, dst, total_len)

这段代码有两个地方值得停下来看。第一是ihl，IP 首部长度是可变的，带选项的包首部会超过 20 字节，如果你固定从第 20 字节开始读传输层，遇到带时间戳选项的包就会解析出一堆乱码，查半天以为是协议识别错了。第二是分片处理，frag_off非零说明这是后续分片，里面根本没有完整的传输层头，硬解只会得到错误端口。常见做法是先把分片单独缓存，等首片到了再按偏移拼，或者干脆在 BPF 里加not ip[6:2] & 0x1fff把分片过滤掉，牺牲一点完整性换稳定性。

传输层解码要顺手把 TCP 的标志位取出来，重传、乱序、连接重置这些指标全靠它。SYN、FIN、RST 三个标志的组合能推断出连接生命周期的阶段，后面算会话时长和失败率都依赖这个。

3.3 流会话重组：五元组哈希表与超时老化

同一个连接的两个方向必须合并成一条会话，否则 TopN 排行里会出现两条各占一半流量的记录，看数据的人会以为链路上有两个高流量源。合并的关键是构造一个方向无关的键：把两端的 IP 和端口按大小排序后拼起来，这样 A→B 和 B→A 落在同一个桶里：

import time class FlowTable: def __init__(self, idle_timeout=60, max_flows=200000): self.flows = {} self.idle_timeout = idle_timeout # 空闲多久判定会话结束 self.max_flows = max_flows # 硬上限，防止内存被打爆 def _key(self, pkt): a = (pkt["src"], pkt["src_port"]) b = (pkt["dst"], pkt["dst_port"]) # 排序保证双向流量落到同一个键 return (pkt["proto"],) + (a + b if a <= b else b + a) def update(self, pkt, now=None): now = now or time.time() key = self._key(pkt) sess = self.flows.get(key) if sess is None: if len(self.flows) >= self.max_flows: self._evict(now) # 先清过期，再考虑淘汰最久未活跃 sess = {"first_ts": now, "last_ts": now, "pkt": 0, "bytes": 0, "fwd": 0, "rev": 0, "app_hint": pkt.get("app_hint")} self.flows[key] = sess sess["last_ts"] = now sess["pkt"] += 1 sess["bytes"] += pkt["len"] if (pkt["src"], pkt["src_port"]) == key[1:3]: sess["fwd"] += pkt["len"] else: sess["rev"] += pkt["len"] return sess def _evict(self, now): dead = [k for k, v in self.flows.items() if now - v["last_ts"] > self.idle_timeout] for k in dead: del self.flows[k]

idle_timeout设多少要看业务。HTTP 短连接多，60 秒足够；长连接推送、数据库连接池这类场景，几十秒的空闲很常见，设太短会把一条连接拆成好几段，统计出的平均会话时长严重偏低。我一般先设 120 秒，观察一段时间会话数的收敛情况再调。max_flows是保命的，没有这个上限，一旦遇到扫描流量或者异常端口探测，哈希表会在几秒内涨到几百万条，进程直接 OOM。

淘汰策略也要注意，别用「满了就清空」这种粗暴做法，那会让统计曲线出现锯齿状的断崖。正确顺序是先按空闲时间清理，还超就按最近活跃时间淘汰最老的 10%，保证内存曲线平滑。

3.4 统计聚合：滑动窗口与 TopN 排序

前端要的是每秒一个点的曲线，不是原始包。所以聚合层维护一个固定长度的环形缓冲，每次快照只算当前窗口内的增量，算完把旧值挤出去：

from collections import deque class WindowCounter: def __init__(self, size=60): # 保留最近 60 个 1 秒窗口，正好画一分钟的曲线 self.windows = deque(maxlen=size) def tick(self, pkt_count, byte_count, flow_table): total_bytes = byte_count self.windows.append({"pps": pkt_count, "bps": total_bytes * 8}) # TopN 按会话累计字节排序，取前 10 top = sorted(flow_table.flows.items(), key=lambda kv: kv[1]["bytes"], reverse=True)[:10] return { "pps": pkt_count, "bps": total_bytes * 8, "top": [{"key": k, "bytes": v["bytes"], "pkts": v["pkt"]} for k, v in top], }

bps乘 8 是从字节换算到比特，这是网络指标的标准口径，别写成字节每秒，否则对着监控面板对不上数会怀疑人生。TopN 的排序每秒钟做一次就够，不要每来一个包排一次，那是纯浪费。如果想看更细的维度，比如按源 IP 聚合而不是按会话聚合，把排序的键换掉就行，聚合层和流表之间只隔着一个字典，改造起来没有耦合。

4. 运行与部署：把源码跑起来的完整步骤

这套东西对环境的依赖集中在两处：抓包权限和 libpcap 库。剩下的都是纯 Python 依赖，装完就能跑。下面按从零到看见图表的顺序走一遍，中间穿插几个容易卡住的位置。

4.1 环境依赖与权限准备

先确认 Python 版本和系统抓包库。Linux 下需要装 libpcap 的开发包，否则绑定库编译不过；Windows 下要装 Npcap 驱动，安装时勾选「支持原始套接字」那一项。依赖清单通常长这样：

# 系统层依赖（Debian/Ubuntu 系） sudo apt-get install -y libpcap-dev python3-dev # Python 依赖 pip install scapy fastapi uvicorn pyyaml

抓包权限是新手第一个翻车点。普通用户默认没有打开网卡的能力，直接跑会报权限错误。两种解法：临时用sudo跑，或者给 Python 解释器加cap_net_raw能力。我更推荐后者，因为sudo跑会把配置文件、日志、缓存全写成 root 属主，后面切回普通用户改文件处处受限：

# 给解释器授予以非 root 身份抓包的能力 sudo setcap cap_net_raw,cap_net_admin=eip $(readlink -f $(which python3))

这条命令执行完，普通用户启动就能抓包。注意升级 Python 之后能力会丢，得重新执行一遍，这个坑我踩过不止一次，表现是「昨天还能跑今天就不行了」。

4.2 配置文件与启动参数

配置集中在一个 YAML 里，把网卡、过滤器、阈值这些易变的项抽出来，代码里不写死。一份能直接用的配置大概是这样：

capture: iface: eth0 # 网卡名，用 ip link 查 bpf: "tcp or udp" # 入口过滤器，越窄越省 CPU snaplen: 128 # 只抓包头，足够了；设 65535 会显著增加拷贝开销 flow: idle_timeout: 120 # 会话空闲超时，秒 max_flows: 200000 # 流表硬上限 metrics: window_seconds: 60 # 曲线窗口长度 top_n: 10 storage: sqlite_path: ./data/metrics.db enable_pcap_ring: false # 需要事后复盘时再开

snaplen这个参数很多人不看就填了最大值。分析流量特征根本用不到载荷内容，128 字节足够覆盖常见协议的头部，把抓取长度截断能明显降低内存拷贝和后续处理压力，还能顺带避免把业务数据落到磁盘上，合规上更省心。

启动入口会按顺序装配各层，先建流表，再建聚合器，然后拉起抓包引擎，最后启动 HTTP 服务：

python main.py --config config/analyzer.yaml

如果只是想验证服务本身通不通，加--dry-run让它读一份内置的样例包序列，不碰网卡，适合在没有权限的环境里先跑通链路。

4.3 实时抓包模式的验证与接口自测

服务起来之后，先别急着开浏览器。用 curl 打一下健康检查接口，确认后端活着，再看指标接口有没有数据：

# 健康检查，返回 status ok 说明服务装配完成 curl -s http://127.0.0.1:8000/api/health # 拉一次当前快照，重点看 pps 是不是在动 curl -s http://127.0.0.1:8000/api/metrics/current | python -m json.tool

如果pps一直是 0，先别怀疑代码。拿 tcpdump 在同一块网卡上抓几个包做交叉验证，它能抓到而你的程序抓不到，问题基本在权限或者过滤器写错；它也抓不到，那就是网卡选错了，或者链路本来就是空的。这个对照法比读日志快得多，应该成为排查第一步。

WebSocket 那条链路可以在浏览器控制台里直接验，也可以在页面打开后看 Network 面板里的帧有没有持续进来。前端图表如果不动但接口有数据，那问题就在推送环节而不是采集环节，边界一下就划清了。

4.4 离线 pcap 回放：没有网卡权限时的调试路径

开发环境常常拿不到抓包权限，这时候回放模式就是后悔药。pcap_reader的接口和抓包引擎完全对齐，都是「给一个包回调」，所以上层解码、流表、聚合一行都不用改，只是把数据源换掉：

from scapy.all import PcapReader def replay(path, on_packet): with PcapReader(path) as reader: for pkt in reader: # 保留原始时间戳，回放时的时序才和真实一致 on_packet(bytes(pkt), ts=float(pkt.time))

注意时间戳要透传。如果回放时用time.time()现取，所有包的时间间隔都会变成处理速度，算出来的 pps 和会话时长全是假的。回放模式还有个好处是结果可复现：同一份 pcap 跑两遍，输出的指标必须完全一致，这正好可以当成回归测试的基线，改完解码器跑一遍对比，改坏了立刻能发现。

5. 排查手册：抓不到包、解析错乱、内存涨满的踩坑记录

这套东西跑通不难，难的是跑稳。下面这几条都是真实遇到过的，按现象、原因、解决的顺序写，遇到相似症状可以直接对号入座。

5.1 抓包层：三条高频翻车记录

现象一：程序启动无报错，但指标一直是零。原因多半是网卡名写错或者权限没生效。容器环境里尤其容易中招，容器看到的网卡名和宿主机完全不是一回事，eth0在容器内可能对应宿主机的某个虚拟网桥。解决方式是先ip link确认容器内的接口列表，再用tcpdump -i 该接口验证有没有流量经过；如果容器内根本看不到物理网卡，需要把网卡以 host 网络模式暴露进来，或者把抓包进程放到宿主机侧跑。

现象二：跑一段时间后丢包计数持续上涨，指标偏低。原因是回调里做的事太重。有人在prn里直接调解码和入库，抓包线程被拖慢，内核缓冲区满了就开始丢。解决办法是把回调改成纯生产者，只往有界队列里塞字节串，解码放在独立线程消费；队列满了宁可丢包也不能阻塞抓包线程，同时把丢弃计数暴露成指标，让自己知道丢了 Compatible 多少。

现象三：BPF 过滤器写了但感觉没生效，CPU 依然很高。常见原因是过滤器语法有拼写错误被静默忽略，或者用了host这类会触发反向解析的表达式。BPF 里host 1.2.3.4在某些实现下会尝试做名称解析，抓包时延会被拖长。改成纯数字条件、加上-n语义的写法更稳，写完用tcpdump -d编译一遍确认语法通过。

5.2 解析与聚合层：两条容易被忽略的坑

现象四：TopN 排行里同一个连接出现两条记录，各占一半流量。原因是五元组键没有做方向归一。修法是构造键时把两端 IP 和端口排序后再拼，让 A→B 和 B→A 落到同一个桶；同时在会话里分别累计正向和反向字节，展示时再决定是合并还是分开。这个坑在处理 NAT 环境时还会叠加一层：同一个内网地址映射到不同外网端口，键里必须带端口，只用 IP 会把完全无关的会话混成一条。

现象五：进程内存线性上涨，几个小时就被系统杀掉。九成是流表没有上限或者超时清理没生效。检查三个点：max_flows有没有配置、_evict是不是真的被调用、时间戳是不是用了回放包的原始时间而不是当前时间。最后这个最隐蔽——回放历史 pcap 时，包时间戳可能是几年前的，按「当前时间减去最后活跃时间」算空闲，所有会话都判定为活跃，永远清不掉，内存只涨不跌。

# 错误示范：用墙钟时间判断回放会话是否过期 idle = time.time() - sess["last_ts"] # 正确做法：用数据源时间推进，抓包和回放共用同一套逻辑 idle = clock.now() - sess["last_ts"]

注意：引入一个统一的时钟抽象，实时模式返回当前时间，回放模式返回当前包的时间戳。这一层抽象能省掉后面无数个「为什么回放和实时结果不一样」的疑问。

6. 二次开发：加一个自定义协议解析器和一套新指标

想把这套东西改成自己业务能用，最常见的两个动作是加协议识别和加统计维度。加协议解析器不要动主流程，在decode/app_probe.py里按「端口特征 + 首字节模式」写一个函数，返回协议名和关键字段，注册到一个列表里就行。比如识别某个内部二进制协议：

def probe_custom(payload: bytes, sport: int, dport: int): # 只对固定端口做探测，避免全量包都进这里增加开销 if 9000 not in (sport, dport) or len(payload) < 8: return None magic = payload[0:4] if magic != b"\xAA\x55\x01\x00": return None # 第 5 字节是消息类型，第 6-8 字节是业务流水号 return {"app": "custom_rpc", "msg_type": payload[4], "seq": int.from_bytes(payload[5:8], "big")} PROBES = [probe_custom] # 新协议注册在这里，主链路不需要改

加新指标同理，在聚合层的窗口计数器里多累加一个字段，再在快照接口的返回结构里带上。这里有个我踩过的点：新字段别忘了同步更新前端的解析逻辑，否则接口多返回了字段前端不认，图表看着没变化，你会以为后端没生效，来回折腾半天。验证方式是先 curl 接口看到字段有值，再看浏览器 Network 面板确认前端拿到了，最后才看渲染，三段分开查比一起猜快得多。

改造前建议先做一件事：拿一份固定的 pcap 当基准，改之前跑一遍存下输出，改之后再跑一遍做 diff。这套流程我后来做任何解析类改动都会走一遍，哪怕只是调整一个偏移量。抓包分析这类代码最坑的地方在于，它错了不会抛异常，只会安安静静给你一份看起来合理的错误数据，等你基于它做出判断才发现方向全反了。从那以后我每次动解码逻辑，都强制先用基准 pcap 跑一遍对比，宁可多花三分钟，也不想再体验一次对着假数据排查一整晚的滋味。希望帮到你。

本文还有配套的精品资源，点击获取