监控网络流量的脚本如何编写

wen 实用脚本 2026-07-12 1

本文目录导读：

监控网络流量的脚本如何编写

使用Python + psutil（推荐）
监控特定网卡流量
Shell脚本方式
高级版：带日志和统计功能
安装依赖
使用建议

使用Python + psutil（推荐）

import psutil
import time
from datetime import datetime
def monitor_network_usage(interval=1):
    """监控网络流量"""
    old_value = psutil.net_io_counters()
    try:
        while True:
            new_value = psutil.net_io_counters()
            # 计算每秒传输的字节数
            sent_per_sec = (new_value.bytes_sent - old_value.bytes_sent) / interval
            recv_per_sec = (new_value.bytes_recv - old_value.bytes_recv) / interval
            # 转换为更易读的格式
            sent = format_bytes(sent_per_sec)
            recv = format_bytes(recv_per_sec)
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            print(f"[{timestamp}] 上传: {sent}/s | 下载: {recv}/s")
            old_value = new_value
            time.sleep(interval)
    except KeyboardInterrupt:
        print("\n监控已停止")
def format_bytes(bytes):
    """将字节转换为人类可读格式"""
    for unit in ['B', 'KB', 'MB', 'GB']:
        if bytes < 1024.0:
            return f"{bytes:.2f} {unit}"
        bytes /= 1024.0
    return f"{bytes:.2f} TB"
if __name__ == "__main__":
    monitor_network_usage()

监控特定网卡流量

import psutil
import time
def monitor_specific_interface(interface_name):
    """监控特定网卡的流量"""
    old_counter = psutil.net_io_counters(pernic=True).get(interface_name)
    if not old_counter:
        print(f"网卡 {interface_name} 不存在")
        return
    print(f"开始监控网卡: {interface_name}")
    print("=" * 50)
    try:
        while True:
            time.sleep(1)
            new_counter = psutil.net_io_counters(pernic=True).get(interface_name)
            if new_counter:
                sent = (new_counter.bytes_sent - old_counter.bytes_sent) / 1024
                recv = (new_counter.bytes_recv - old_counter.bytes_recv) / 1024
                print(f"上传: {sent:.2f} KB/s | 下载: {recv:.2f} KB/s")
                old_counter = new_counter
    except KeyboardInterrupt:
        print("\n监控结束")
# 获取可用网卡列表
def list_interfaces():
    interfaces = psutil.net_io_counters(pernic=True)
    return list(interfaces.keys())
# 使用示例
if __name__ == "__main__":
    interfaces = list_interfaces()
    print("可用网卡:", interfaces)
    if interfaces:
        monitor_specific_interface(interfaces[0])  # 监控第一个网卡

Shell脚本方式

#!/bin/bash
# 简单的网络流量监控脚本
INTERVAL=1
# 获取初始值
OLD_RX=$(cat /proc/net/dev | grep eth0 | awk '{print $2}')
OLD_TX=$(cat /proc/net/dev | grep eth0 | awk '{print $10}')
while true; do
    sleep $INTERVAL
    NEW_RX=$(cat /proc/net/dev | grep eth0 | awk '{print $2}')
    NEW_TX=$(cat /proc/net/dev | grep eth0 | awk '{print $10}')
    RX_RATE=$(( ($NEW_RX - $OLD_RX) / $INTERVAL ))
    TX_RATE=$(( ($NEW_TX - $OLD_TX) / $INTERVAL ))
    # 转换为KB
    RX_KB=$(echo "scale=2; $RX_RATE / 1024" | bc)
    TX_KB=$(echo "scale=2; $TX_RATE / 1024" | bc)
    echo "$(date '+%Y-%m-%d %H:%M:%S') 下载: ${RX_KB}KB/s 上传: ${TX_KB}KB/s"
    OLD_RX=$NEW_RX
    OLD_TX=$NEW_TX
done

高级版：带日志和统计功能

import psutil
import time
import json
import csv
from datetime import datetime, timedelta
from collections import deque
class NetworkMonitor:
    def __init__(self, log_file="network_traffic.csv", history_size=60):
        self.log_file = log_file
        self.history = deque(maxlen=history_size)
        self.init_log_file()
    def init_log_file(self):
        """初始化日志文件"""
        try:
            with open(self.log_file, 'w', newline='') as f:
                writer = csv.writer(f)
                writer.writerow(['timestamp', 'sent_bytes', 'recv_bytes', 'sent_rate', 'recv_rate'])
        except:
            print(f"无法创建日志文件: {self.log_file}")
    def log_traffic(self, data):
        """记录流量数据"""
        try:
            with open(self.log_file, 'a', newline='') as f:
                writer = csv.writer(f)
                writer.writerow([
                    data['timestamp'],
                    data['total_sent'],
                    data['total_recv'],
                    data['sent_rate'],
                    data['recv_rate']
                ])
        except:
            pass
    def get_statistics(self):
        """获取统计信息"""
        if len(self.history) < 2:
            return None
        rates = [h['recv_rate'] for h in self.history if h['recv_rate'] > 0]
        if rates:
            return {
                'avg_recv': sum(rates) / len(rates),
                'max_recv': max(rates),
                'min_recv': min(rates),
                'total_records': len(self.history)
            }
        return None
    def start_monitoring(self, interval=1):
        """开始监控"""
        print("网络流量监控开始...")
        print("=" * 60)
        old_counters = psutil.net_io_counters()
        try:
            while True:
                time.sleep(interval)
                new_counters = psutil.net_io_counters()
                # 计算速率
                sent_rate = (new_counters.bytes_sent - old_counters.bytes_sent) / interval
                recv_rate = (new_counters.bytes_recv - old_counters.bytes_recv) / interval
                timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                # 准备数据
                data = {
                    'timestamp': timestamp,
                    'total_sent': new_counters.bytes_sent,
                    'total_recv': new_counters.bytes_recv,
                    'sent_rate': sent_rate,
                    'recv_rate': recv_rate
                }
                # 添加到历史记录
                self.history.append(data)
                # 记录日志
                self.log_traffic(data)
                # 显示信息
                print(f"[{timestamp}]", end=" ")
                print(f"上传: {self._format(sent_rate)}/s | 下载: {self._format(recv_rate)}/s", end=" ")
                # 显示统计信息
                stats = self.get_statistics()
                if stats:
                    print(f"| 平均下载: {self._format(stats['avg_recv'])}/s", end="")
                print()
                old_counters = new_counters
        except KeyboardInterrupt:
            print("\n监控已停止")
            self.show_summary()
    def _format(self, bytes_per_sec):
        """格式化显示"""
        for unit in ['B', 'KB', 'MB', 'GB']:
            if bytes_per_sec < 1024:
                return f"{bytes_per_sec:.2f} {unit}"
            bytes_per_sec /= 1024
        return f"{bytes_per_sec:.2f} TB"
    def show_summary(self):
        """显示统计摘要"""
        print("\n" + "=" * 60)
        print("监控摘要:")
        if self.history:
            total_sent = self.history[-1]['total_sent'] - self.history[0]['total_sent']
            total_recv = self.history[-1]['total_recv'] - self.history[0]['total_recv']
            print(f"总上传量: {self._format(total_sent)}")
            print(f"总下载量: {self._format(total_recv)}")
            print(f"监控时长: {len(self.history)} 秒")
            print(f"日志文件: {self.log_file}")
if __name__ == "__main__":
    monitor = NetworkMonitor()
    monitor.start_monitoring(interval=2)  # 每2秒更新一次

安装依赖

# Python方案需要安装psutil
pip install psutil
# 如果使用虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows
pip install psutil

使用建议

选择合适的工具：
- 简单监控：使用Shell脚本
- 需要统计分析：使用Python方案
- 长期记录：使用带日志功能的版本
性能考虑：
- 监控间隔不要太小（建议1-5秒）
- 避免频繁写日志（可以批量写入）
扩展功能：
- 添加告警功能（流量超过阈值时通知）
- 支持多网卡同时监控
- 添加Web界面显示

需要根据具体需求选择合适的脚本版本。