本文目录导读：
使用 Python + scapy
from scapy.all import *
from collections import defaultdict
import time
class NetworkMonitor:
def __init__(self):
self.traffic_stats = defaultdict(lambda: {'bytes': 0, 'packets': 0})
def packet_callback(self, packet):
if IP in packet:
src = packet[IP].src
dst = packet[IP].dst
size = len(packet)
# 统计源IP流量
self.traffic_stats[src]['bytes'] += size
self.traffic_stats[src]['packets'] += 1
# 统计目标IP流量
self.traffic_stats[dst]['bytes'] += size
self.traffic_stats[dst]['packets'] += 1
def print_stats(self):
print(f"\n{time.strftime('%Y-%m-%d %H:%M:%S')}")
print("IP流量统计（Top 10）：")
sorted_ip = sorted(
self.traffic_stats.items(),
key=lambda x: x[1]['bytes'],
reverse=True
)[:10]
for ip, stats in sorted_ip:
print(f"{ip:20s} 包数: {stats['packets']:6d} "
f"流量: {stats['bytes']/1024:.2f} KB")
def monitor(self, interface='eth0', duration=60):
print(f"开始监控网卡 {interface}，持续 {duration} 秒...")
sniff(
iface=interface,
prn=self.packet_callback,
store=0,
timeout=duration
)
self.print_stats()
# 使用示例
if __name__ == "__main__":
monitor = NetworkMonitor()
monitor.monitor(interface='eth0', duration=30)
使用 Bash + tcpdump
#!/bin/bash
# 网络流量监控脚本
INTERFACE="eth0"
DURATION=60
LOG_FILE="/var/log/network_traffic.log"
monitor_traffic() {
echo "=== 监控开始: $(date) ===" >> $LOG_FILE
# 使用tcpdump捕获流量
tcpdump -i $INTERFACE -n -t 2>/dev/null | while read line; do
# 提取源IP和目标IP（假设是IP包）
if [[ $line =~ IP ]]; then
src_ip=$(echo $line | awk '{print $3}' | cut -d'.' -f1-4)
dst_ip=$(echo $line | awk '{print $5}' | cut -d'.' -f1-4)
size=$(echo $line | awk '{print $NF}' | tr -d '()')
echo "$(date +%H:%M:%S) $src_ip -> $dst_ip (大小: $size bytes)" >> $LOG_FILE
fi
done &
# 每秒显示流量统计
while [ $SECONDS -lt $DURATION ]; do
# 使用/proc/net/dev查看接口流量
rx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $2}')
tx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $10}')
echo "时间: $(date +%H:%M:%S) RX: $((rx_bytes/1024)) KB TX: $((tx_bytes/1024)) KB"
sleep 1
done
echo "=== 监控结束: $(date) ===" >> $LOG_FILE
}
# 实时流量监控
realtime_monitor() {
echo "实时流量监控（按Ctrl+C退出）"
while true; do
rx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $2}')
tx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $10}')
# 计算速度（每秒）
if [ -n "$last_rx" ]; then
rx_speed=$(( ($rx_bytes - $last_rx) / 1024 ))
tx_speed=$(( ($tx_bytes - $last_tx) / 1024 ))
echo "下载: ${rx_speed}KB/s 上传: ${tx_speed}KB/s"
fi
last_rx=$rx_bytes
last_tx=$tx_bytes
sleep 1
done
}
# 主菜单
case "$1" in
monitor)
monitor_traffic
;;
realtime)
realtime_monitor
;;
*)
echo "用法: $0 {monitor|realtime}"
exit 1
;;
esac
使用 Python + psutil (更简单)
#!/usr/bin/env python3
import psutil
import time
from datetime import datetime
class SimpleNetworkMonitor:
def __init__(self, interface='eth0'):
self.interface = interface
self.last_bytes = None
self.last_time = None
def get_network_stats(self):
net_io = psutil.net_io_counters(pernic=True)
if self.interface in net_io:
return net_io[self.interface]
return None
def monitor_speed(self, duration=60):
print(f"监控 {self.interface} 网卡流量速度...")
# 获取初始值
init_stats = self.get_network_stats()
if not init_stats:
print(f"未找到网卡 {self.interface}")
return
last_sent = init_stats.bytes_sent
last_recv = init_stats.bytes_recv
last_time = time.time()
while duration > 0:
time.sleep(1)
current_stats = self.get_network_stats()
if not current_stats:
break
now = time.time()
time_diff = now - last_time
# 计算速度
sent_speed = (current_stats.bytes_sent - last_sent) / time_diff / 1024 # KB/s
recv_speed = (current_stats.bytes_recv - last_recv) / time_diff / 1024 # KB/s
print(f"\r{datetime.now().strftime('%H:%M:%S')} "
f"↑ {sent_speed:.1f} KB/s ↓ {recv_speed:.1f} KB/s", end='')
last_sent = current_stats.bytes_sent
last_recv = current_stats.bytes_recv
last_time = now
duration -= 1
print("\n监控结束")
# 使用示例
if __name__ == "__main__":
monitor = SimpleNetworkMonitor('eth0')
monitor.monitor_speed(duration=30)
高级用法：nethogs 风格的进程监控
#!/usr/bin/env python3
import psutil
import time
from collections import defaultdict
class ProcessNetworkMonitor:
def __init__(self):
self.process_traffic = defaultdict(lambda: {'sent': 0, 'recv': 0})
def get_process_connections(self):
connections = psutil.net_connections()
process_data = {}
for conn in connections:
if conn.status == 'ESTABLISHED' and conn.pid:
try:
process = psutil.Process(conn.pid)
process_name = process.name()
if process_name not in process_data:
process_data[process_name] = {
'pid': conn.pid,
'connections': []
}
process_data[process_name]['connections'].append({
'local': conn.laddr,
'remote': conn.raddr if conn.raddr else None
})
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return process_data
def monitor_process_traffic(self, duration=30):
print("监控进程网络流量...")
print(f"{'进程名称':20s} {'PID':8s} {'上传':>10s} {'下载':>10s} {'连接数':>8s}")
print("-" * 56)
while duration > 0:
process_data = self.get_process_connections()
for process_name, data in process_data.items():
# 简单统计
num_connections = len(data['connections'])
sent = self.process_traffic[process_name]['sent']
recv = self.process_traffic[process_name]['recv']
# 更新流量（实际项目需要更复杂的计算）
sent += 1024 # 示例数据
recv += 2048 # 示例数据
print(f"{process_name:20s} {str(data['pid']):8s} "
f"{sent/1024:>8.1f}KB {recv/1024:>8.1f}KB "
f"{num_connections:>8d}")
self.cleanup_old_data()
time.sleep(2)
duration -= 2
# 清屏
print("\033[2J\033[H", end='')
def cleanup_old_data(self):
# 清理超过时间的进程数据
pass
# 使用示例
if __name__ == "__main__":
monitor = ProcessNetworkMonitor()
monitor.monitor_process_traffic(duration=30)
安装依赖
# Python依赖 pip install scapy psutil # 系统工具（Ubuntu/Debian） sudo apt-get install tcpdump nethogs iftop
使用权限
大多数网络监控需要 root 权限：
# 使用sudo运行脚本 sudo python3 network_monitor.py sudo bash network_monitor.sh
注意事项
- 性能影响：持续监控会消耗系统资源
- 隐私问题：监控他人网络流量可能涉及隐私
- 权限要求：需要适当的系统权限
- 数据存储：大量流量数据需要合理存储和处理
选择哪种方法取决于你的具体需求：
- 简单监控：使用 psutil
- 详细分析：使用 scapy
- 命令行工具：使用 tcpdump/nethogs
- 进程级监控：使用 psutil + netstat
