简介：这份网络流量分析器源码面向从事网络监控、安全运维与VC++开发的技术人员，提供一套可编译运行的抓包流量分析程序，可用于业务流量监控、病毒木马后门流量监控、ARP病毒监控及网络异常监控等实际场景。压缩包共36个文件，约168KB，以h头文件与cpp源文件为主体，配合rc资源脚本、ico与bmp图标位图、mdb与db数据文件、sln与vcxproj工程文件及exe可执行程序，构成完整的VC++工程结构。程序实现单点流量、点到点流量与协议流量排名分析，支持按流量统计和按流量占比统计，可显示基于协议和目标端口的累计流量，并提示ARP流量占比、开展ARP流量分析。源码涉及内存表扫描、BarChart图表控件使用等关键技术点，读者可据此理解抓包记录的组织方式、统计逻辑与图表呈现方法，并在此基础上扩展监控指标或移植到自身业务环境。目前已有911人学习下载。

1. 抓包抓不到重点？这套网络流量分析器源码把链路拆开给你看

线上接口突然变慢，日志里全是正常返回，服务端 CPU 也不高，最后发现是某个下游连接在反复重传。这种问题靠打日志基本无解，得看链路上真正跑了什么。这套网络流量分析器源码解决的就是这件事：从网卡抓包、按协议解析、统计会话、输出可读的分析结果，一条链路走完。它适合两类人——想搞懂抓包解析全流程的后端和运维，以及要拿现成骨架做流量监控、协议识别、异常检测的开发者。源码结构不复杂，但把抓包、解码、统计、展示这几层切得比较清楚，拿来当网络方向的学习底子或者二次开发起点都合适。下面按“能干什么、怎么跑、怎么改”的顺序拆。

2. 架构与目录：抓包、解码、统计、输出四层怎么串起来

2.1 整体分层与数据流

这套源码的核心思路是流水线：抓包层负责从网卡或离线文件拿到原始字节，解码层把字节按以太网帧、IP 包、TCP/UDP 段逐层剥开，统计层按五元组聚合会话并累计字节数、包数、时长，输出层把结果打成表格或写文件。四层之间用统一的数据结构传递，解码层不关心数据从哪来，统计层不关心字节怎么解，这样离线分析和实时抓包能复用同一套逻辑。

常见做法是定义一个Packet结构体，字段包括时间戳、源/目的 IP、源/目的端口、协议号、载荷长度、TCP 标志位。抓包层填时间戳和原始字节，解码层填其余字段，统计层只读这个结构。这样分层的好处是加新协议解析时只动解码层，换输出格式时只动输出层，互不影响。

# 统一的数据载体，四层之间只传这个结构 class Packet: def __init__(self): self.ts = 0.0 # 抓包时间戳，秒 self.src_ip = "" # 源 IP self.dst_ip = "" # 目的 IP self.src_port = 0 # 源端口 self.dst_port = 0 # 目的端口 self.proto = 0 # 协议号：6=TCP, 17=UDP self.payload_len = 0 # 载荷长度 self.tcp_flags = 0 # TCP 标志位，用于识别 SYN/FIN/RST

这段结构体是整个项目的“通用货币”。ts用浮点秒而不是整数毫秒，是为了在统计会话时长时保留精度；tcp_flags单独留字段，是因为后面判断连接建立和断开要靠 SYN、FIN、RST，如果只存协议号就没法区分一次正常关闭和一次异常重置。实际写的时候我一般会再加一个raw字段存原始字节，方便调试时回看，但统计层不读它，避免内存占用过大。

2.2 目录结构与模块职责

源码目录大致是这样组织的，不同版本可能略有出入，但分层逻辑一致：

traffic-analyzer/ ├── capture/ # 抓包层：网卡抓包、pcap 文件读取 │ ├── live.py # 实时抓包 │ └── offline.py # 离线 pcap 读取 ├── decode/ # 解码层：以太网/IP/TCP/UDP 解析 │ ├── ethernet.py │ ├── ip.py │ └── transport.py ├── stats/ # 统计层：会话聚合、流量统计 │ └── session.py ├── output/ # 输出层：表格、文件、JSON │ └── report.py ├── config.py # 过滤规则、抓包参数 └── main.py # 入口，串起四层

capture/下实时和离线分开，是因为两者拿数据的方式不同：实时抓包要绑定网卡、设混杂模式，离线读取只是按顺序读文件。但两者对外都暴露同一个迭代器接口，main.py里换一行配置就能切换。decode/按协议层拆文件，是为了每层只处理自己那一段，以太网层剥掉 14 字节头交给 IP 层，IP 层剥掉 20 字节头交给传输层，职责清晰。stats/只有一个会话聚合模块，因为统计逻辑相对集中。output/独立出来，方便后面加新的展示形式。

2.3 入口怎么把四层串起来

main.py的逻辑很短，就是取配置、建抓包源、循环解码、喂给统计、最后输出。关键在于循环里对异常包的处理：解码失败不能整个程序崩掉，要跳过继续。

from capture.live import LiveCapture from capture.offline import OfflineCapture from decode.ethernet import decode_ethernet from stats.session import SessionTable from output.report import print_report import config def build_source(cfg): # 根据配置决定实时抓包还是读离线文件 if cfg.mode == "live": return LiveCapture(cfg.iface, cfg.filter) return OfflineCapture(cfg.pcap_path) def run(): cfg = config.load() source = build_source(cfg) table = SessionTable() for raw in source: try: pkt = decode_ethernet(raw) # 逐层解码，返回 Packet except ValueError: continue # 坏包跳过，不中断 if pkt is None: continue # 非 IP 包直接忽略 table.add(pkt) # 按五元组聚合 print_report(table, top_n=cfg.top_n)

build_source把抓包源的创建收在一处，cfg.mode是live还是offline决定走哪个分支，这样切换分析方式不用改主循环。decode_ethernet返回None表示这个包不是 IP 流量（比如 ARP），直接忽略；抛ValueError表示字节不够或格式不对，跳过。table.add是统计层的入口，内部按五元组找已有会话或新建。print_report的top_n控制输出前几名，避免终端刷屏。这个入口没有花哨设计，但把“坏包不崩、非目标流量不统计”这两个实际场景处理了，跑真实流量时很关键。

3. 跑起来：环境、抓包权限和第一份分析结果

3.1 依赖安装与配置

这套源码用 Python 写，抓包底层依赖 libpcap，Python 侧常见做法是用 scapy 或 pcapy 这类库。按项目里requirements.txt装依赖即可，核心就两三个包。装之前确认系统有 libpcap 开发库，否则底层库编译会失败。

# 系统层依赖，Debian/Ubuntu 系 sudo apt-get install -y libpcap-dev # Python 依赖 pip install -r requirements.txt

libpcap-dev提供抓包的系统调用接口，Python 的抓包库只是它的封装，缺了它pip install阶段就会报找不到头文件。requirements.txt里一般会有抓包库、解析库和表格输出库，版本不用刻意锁死，但抓包库和系统 libpcap 版本差太多时会有兼容问题，遇到报错先看这一层。

配置文件config.py里几个参数要按场景改：

iface用ip addr或ifconfig查实际网卡名，不同系统命名不一样，填错会直接报“找不到设备”。filter是 BPF 语法，写tcp port 80就只抓 80 端口的 TCP，能大幅减少无关流量，跑大流量时必设。

3.2 抓包权限这个坎

实时抓包要读网卡原始数据，普通用户没权限，直接跑会报PermissionError或底层返回空。常见做法是用sudo跑，或者给 Python 解释器加抓包能力。

# 方式一：直接提权运行 sudo python main.py # 方式二：给解释器加能力，之后免 sudo sudo setcap cap_net_raw,cap_net_admin=eip $(readlink -f $(which python3))

方式一简单但每次都要 sudo，调试时环境变量和路径容易和普通用户不一致。方式二给解释器加了原始套接字和管理网卡的能力，之后普通用户也能抓，适合反复调试。注意方式二改的是解释器权限，换虚拟环境后要重新设。如果只是分析已有的 pcap 文件，用offline模式，完全不需要这些权限，新手建议先从这个模式入手。

3.3 用离线文件跑通第一遍

手上没有现成流量时，先拿一个 pcap 文件跑离线模式，确认整条链路通。把config.py里mode改成offline，pcap_path指向文件，然后运行。

python main.py

正常输出是按会话聚合的表格，每行一个五元组，列出包数、字节数、持续时间。第一次跑建议先看三件事：总包数对不对、有没有明显解析失败的计数、前几名会话是不是符合预期。如果总包数是 0，先查pcap_path和文件格式；如果解析失败计数很高，多半是链路层类型不匹配（比如文件是 Linux cooked 格式而不是以太网帧），需要在解码层加分支。跑通离线再切实时，能少踩很多权限和网卡的坑。

4. 核心代码拆解：解码层和会话统计怎么写

4.1 以太网到 IP 的逐层剥离

解码层是整个项目最需要小心的地方，因为网络字节序、可变头长度、分片这些细节都在这里。以太网帧头固定 14 字节，其中前 12 字节是源/目的 MAC，后 2 字节是上层协议类型，0x0800表示 IPv4。剥掉这 14 字节，剩下的交给 IP 层。

import struct ETH_HEADER_LEN = 14 ETH_TYPE_IPV4 = 0x0800 def decode_ethernet(raw): if len(raw) < ETH_HEADER_LEN: raise ValueError("frame too short") # 前 12 字节 MAC 这里不解析，直接跳过 eth_type = struct.unpack("!H", raw[12:14])[0] if eth_type != ETH_TYPE_IPV4: return None # 非 IPv4，忽略 return decode_ip(raw[ETH_HEADER_LEN:])

struct.unpack("!H", ...)里的!表示网络字节序（大端），这是网络协议解析最容易翻车的地方——x86 是小端，不显式指定就会把0x0800读成0x0008，然后所有包都被当成未知协议。raw[12:14]取的是以太网类型字段，判断不等于 IPv4 就返回None，让上层跳过。这里没解析 MAC 地址，因为统计层用不到，需要时再加。实际抓包里还有 VLAN 标签（0x8100），带标签的帧头会多 4 字节，遇到时要在这一步先判断再偏移，否则后面全错位。

4.2 IP 头解析与可变长度处理

IP 头不是固定 20 字节，IHL字段表示头长度（以 4 字节为单位），有选项时会更长。解析时必须按IHL算偏移，写死 20 字节在带选项的包上会解析出错。

def decode_ip(data): if len(data) < 20: raise ValueError("ip header too short") ver_ihl = data[0] version = ver_ihl >> 4 ihl = (ver_ihl & 0x0F) * 4 # 头长度，单位字节 if version != 4: return None proto = data[9] # 协议号 src_ip = ".".join(str(b) for b in data[12:16]) dst_ip = ".".join(str(b) for b in data[16:20]) total_len = struct.unpack("!H", data[2:4])[0] payload = data[ihl:total_len] # 按总长度截，去掉以太网填充 return decode_transport(proto, src_ip, dst_ip, payload)

ver_ihl >> 4取高 4 位是版本号，& 0x0F取低 4 位是头长度，再乘 4 得到字节数，这是 IP 头解析的标准操作。data[9]是协议号，6 是 TCP、17 是 UDP。源/目的 IP 各取 4 字节转成点分十进制。total_len从偏移 2 取 2 字节，用它截payload而不是直接取到末尾，是因为以太网帧有最小长度限制，小包后面会填充无用字节，不按总长度截会把填充当载荷，统计字节数就偏大。这个细节不注意，小包多的场景统计结果会明显虚高。

4.3 会话聚合：五元组和超时淘汰

统计层的核心是用五元组（源 IP、源端口、目的 IP、目的端口、协议）做键，把同一个连接的所有包累加到一起。难点在于连接什么时候算结束：TCP 有 FIN/RST 可以判断，UDP 没有连接概念，只能靠超时。

import time class SessionTable: def __init__(self, timeout=60): self.sessions = {} # key: 五元组, value: 统计信息 self.timeout = timeout # 空闲超时秒数 def _key(self, pkt): # 双向流量归一化：小端口在前，保证请求和响应落到同一会话 a = (pkt.src_ip, pkt.src_port) b = (pkt.dst_ip, pkt.dst_port) pair = tuple(sorted([a, b])) return (pair[0], pair[1], pkt.proto) def add(self, pkt): key = self._key(pkt) now = pkt.ts s = self.sessions.get(key) if s is None: s = {"packets": 0, "bytes": 0, "start": now, "last": now} self.sessions[key] = s s["packets"] += 1 s["bytes"] += pkt.payload_len s["last"] = now # TCP 收到 FIN 或 RST，标记会话可关闭 if pkt.proto == 6 and (pkt.tcp_flags & 0x01 or pkt.tcp_flags & 0x04): s["closed"] = True

_key里对两端排序是关键设计：同一个连接的两个方向，源和目的会互换，不归一化就会变成两个会话，统计翻倍。sorted按 IP 和端口排序，保证两个方向算出同一个键。add里累加包数和字节数，更新最后活跃时间。TCP 的 FIN 标志是0x01、RST 是0x04，收到就标记closed，输出时可以区分正常关闭和异常重置。timeout字段留给清理逻辑用：长时间没流量的会话要淘汰，否则跑久了内存一直涨。清理一般在主循环里定期扫一遍，把now - last > timeout的删掉。UDP 会话只能靠这个超时判断结束，超时设太短会把长连接拆成多个，设太长内存回收慢，60 秒是个折中值，按业务调。

4.4 输出层：把统计结果打成能看的表

输出层不复杂，但决定了结果好不好读。常见做法是按字节数降序排，取前 N 名，每行列出五元组、包数、字节数、持续时长。

def print_report(table, top_n=10): rows = [] for key, s in table.sessions.items(): (sip, sport), (dip, dport), proto = key duration = s["last"] - s["start"] rows.append((s["bytes"], sip, sport, dip, dport, proto, s["packets"], duration)) rows.sort(reverse=True) # 按字节数降序 print(f"{'源':<22}{'目的':<22}{'协议':<6}{'包数':<8}{'字节':<10}{'时长'}") for r in rows[:top_n]: src = f"{r[1]}:{r[2]}" dst = f"{r[3]}:{r[4]}" print(f"{src:<22}{dst:<22}{r[5]:<6}{r[6]:<8}{r[0]:<10}{r[7]:.1f}s")

rows.sort(reverse=True)直接按元组第一个元素（字节数）降序，因为元组比较从第一项开始，字节数相同才比后面的，实际场景够用。格式化里:<22是左对齐占 22 字符，保证列对齐，IP 加端口最长差不多这个宽度。duration用最后减开始，反映会话活跃跨度。这个输出适合终端快速看，要长期留存或做进一步分析，把print换成写 CSV 或 JSON 就行，统计层的数据结构不用动，这也是分层的好处。

5. 避坑与排查：跑真实流量时最容易翻车的几处

5.1 抓到的包全是 0 或者只有自己的流量

现象是实时模式跑起来，统计结果里只有本机发起的连接，看不到其他机器的流量。原因是网卡没进混杂模式，默认只收发给自己的帧。解决是在打开抓包设备时显式设置混杂模式，抓包库一般有对应参数，比如promisc=True。另外虚拟机环境里网卡可能是 NAT 模式，宿主机和其他虚拟机的流量根本到不了这块虚拟网卡，这种情况要么换桥接模式，要么在网关侧抓。先确认抓包位置对不对，再怀疑代码。

5.2 字节数统计明显偏大

现象是统计出来的总字节数比实际流量大不少，小包多的场景尤其明显。原因是 IP 头解析时没按total_len截断，把以太网帧的填充字节也算进了载荷。以太网最小帧 64 字节，小包后面会补零到最小长度，不截断就会多算。解决是按 IP 头里的总长度字段截payload，代码里data[ihl:total_len]就是干这个的。改完再对比一次抓包工具的总字节数，能对上就说明修对了。

5.3 同一连接被统计成两条

现象是请求和响应分别出现在两行，包数和字节数各算一半。原因是会话键没做双向归一化，源和目的互换后算出了不同的键。解决是在生成键时对两端排序，保证A→B和B→A落到同一个键。注意排序要按 IP 和端口一起排，只排 IP 不排端口，同一对 IP 上的多个连接会混在一起。改完后看输出，一个连接的上下行应该合并成一行。

5.4 跑一段时间内存持续上涨

现象是长时间抓包后进程内存越来越大，最后被系统杀掉。原因是会话表只增不删，短连接和扫描流量会不断产生新键。解决是加超时淘汰，定期扫描把空闲超过阈值的会话删掉，TCP 已关闭的会话可以更早清理。清理频率不用太高，几十秒一次就够，太频繁反而增加开销。另外raw字段如果存了原始字节，统计完要及时释放，别一直挂在会话上。

5.5 离线文件解析全失败

现象是离线模式跑起来解析失败计数等于总包数，一条有效会话都没有。原因是链路层类型不匹配，pcap 文件可能不是标准以太网帧，而是 Linux cooked 格式（SLL）或原始 IP。解决是先读 pcap 全局头里的链路类型字段，按类型选择对应的解码入口，SLL格式没有 MAC 头，偏移量不一样。抓包工具导出时也可以指定格式，统一成以太网帧最省事。遇到这种问题先打印前几个包的十六进制，看头部结构就能判断。

6. 二次开发：加协议识别和异常检测的落点

拿到这套源码后，最常见的改造方向是加协议识别和异常检测。协议识别不用改抓包和统计层，在解码层拿到端口和载荷后加一个判断函数就行：按知名端口先做粗分类（80/443 是 Web、22 是 SSH），端口不明确的再看载荷特征。异常检测则挂在统计层之后，对会话表做一轮扫描，比如单会话字节数超过阈值、短时间内大量 SYN 没有对应 ACK（扫描特征）、某个目的端口连接数异常集中，都可以作为规则。这些规则读的都是现成的会话结构，不用动底层。

改造时我建议先加一个可配置的规则表，把阈值和开关放配置里，别写死在代码里。验证方法很直接：拿一份已知包含异常的 pcap 跑一遍，看规则能不能命中，再拿正常流量跑一遍看误报率。下面是一个规则扫描的骨架，挂在print_report之前：

def detect_anomalies(table, rules): alerts = [] for key, s in table.sessions.items(): (sip, sport), (dip, dport), proto = key # 规则一：单会话流量过大 if s["bytes"] > rules["max_session_bytes"]: alerts.append(("大流量会话", sip, dip, s["bytes"])) # 规则二：短时间大量包，疑似扫描或攻击 duration = max(s["last"] - s["start"], 0.001) if s["packets"] / duration > rules["max_pps"]: alerts.append(("高速率会话", sip, dip, s["packets"])) return alerts

rules从配置读，max_session_bytes和max_pps按实际网络规模调，内网和公网阈值差很多。duration取max(..., 0.001)是防止除零，单个包的会话时长为 0 会直接崩。这个骨架只做了两条规则，加规则就是往循环里加判断，结构不用变。验证时把告警和人工分析结果对一遍，误报多的规则先调阈值再考虑删。

血泪经验是：改解码层一定要拿真实 pcap 回归，别只用手造的包测。我见过按标准文档写的解析在真实流量上因为 VLAN 标签和 IP 选项直接错位，手造包永远测不出来。从那以后我每次动解码逻辑，都强制拿几份不同来源的 pcap 各跑一遍，对比总包数和字节数再提交。希望帮到你。

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