A:更新部分代码和文档

This commit is contained in:
jimmy committed 2025-08-15 08:11:35 +08:00
1 parent 16bb899348
commit 04bb8099ac
12 files changed
+2835 -162

No files matched your search

+261 -4
View File
@@ -1,11 +1,268 @@
# 项目介绍 # NetVizor 网络管理系统
这是一个网络管理项目,它用的是C#,网络信息使用ETW,并集成防火墙API,提供HTTP和WebSocket接口来实现数据的传递和接收,使用Sqlite作为底层数据库。使用Vue作为前端并通过WebView2套壳实现主页面渲染
这是一个基于C#开发的高级网络监控管理系统,通过ETW(Event Tracing for Windows)技术实现深度网络流量监控,集成Windows防火墙管理功能,提供HTTP和WebSocket接口对外提供数据服务,使用SQLite作为底层数据库存储,前端采用Vue.js并通过WebView2实现桌面应用界面。
我所有的环境并非全部在默认位置安装,比如部分环境如 Git 之类的在 D:\Env 下 ## 系统概述
NetVizor是一个功能丰富的网络监控和管理工具,主要用于:
- 实时网络流量监控和分析
- 应用程序网络行为跟踪
- Windows防火墙规则管理
- 网络连接状态监控
- 流量统计和数据可视化
你可以使用TODO.md 文档来管理你的任务,或者记录任何东西,里面的内容完全由你管理,我不会干预,如果有内容你可以直接清空,它就像你的草稿纸。 ## 技术架构
### 后端架构(C#/.NET)
- **框架版本**: .NET 8.0
- **架构模式**: 分层架构 + 模块化设计
- **核心技术**: ETW、Windows API、COM接口、WebSocket、HTTP服务
### 前端技术栈
- **框架**: Vue.js 3
- **渲染引擎**: Microsoft WebView2
- **UI组件**: 自定义组件库
- **数据可视化**: Chart.js集成
### 数据存储
- **主数据库**: SQLite 3
- **ORM**: Dapper
- **数据结构**: 关系型数据库设计,支持分时聚合
## 项目结构详解
### 核心模块
#### 1. Shell 模块(主程序)
**位置**: `Shell/`
**功能**: 应用程序启动入口和主界面
- **App.xaml.cs**: 应用程序启动逻辑,全局异常处理,服务初始化
- **Views/**: WPF视图文件(主窗口、设置窗口、分析窗口等)
- **UserControls/**: 自定义用户控件
- **Services/**: 业务服务层
- **Utils/**: 界面相关工具类
- **wwwroot/**: Vue前端静态资源文件
#### 2. Data 模块(数据访问层)
**位置**: `Data/`
**功能**: 数据库操作和数据模型定义
- **NetVizorDbContext.cs**: 数据库上下文,表结构定义
- **Models/**: 数据模型类(AppInfo、AppNetwork、GlobalNetwork、AppSetting)
- **Repositories/**: 数据访问仓储类
- **Services/**: 数据服务层
- **Core/**: 核心数据管理类
**数据表结构**:
- `AppInfo`: 应用程序信息表
- `AppNetwork`: 应用网络记录表
- `GlobalNetwork`: 全局网络流量表
- `AppSetting`: 应用设置表
- 聚合表: 按小时/天/周/月的流量统计表
#### 3. Utils.ETW 模块(网络监控核心)
**位置**: `Utils.ETW/`
**功能**: 基于ETW的网络事件捕获和处理
- **Core/ETWNetworkCapture.cs**: ETW网络事件捕获器
- **Etw/EnhancedEtwNetworkManager.cs**: 增强型网络管理器
- **Models/**: 网络事件数据模型
- **Services/**: 网络监控服务
**监控能力**:
- TCP连接监控(建立、断开、发送、接收)
- UDP包监控
- DNS查询监控
- HTTP请求监控
- 进程网络行为关联
#### 4. Utils.Firewall 模块(防火墙管理)
**位置**: `Utils.Firewall/`
**功能**: Windows防火墙规则管理
- **WindowsFirewallApi.cs**: Windows防火墙COM API封装
- **FirewallService.cs**: 防火墙服务接口实现
- **FirewallModel.cs**: 防火墙规则模型定义
**管理功能**:
- 防火墙规则CRUD操作
- 防火墙状态管理
- 规则导入导出
- 批量操作支持
- 规则统计分析
#### 5. Common.Net 模块(网络通信)
**位置**: `Common.Net/`
**功能**: HTTP和WebSocket服务器实现
- **HttpConn/EmbeddedWebServer.cs**: 嵌入式HTTP服务器
- **WebSocketConn/WebSocketManager.cs**: WebSocket管理器
- **Models/**: 通信数据模型
**服务特性**:
- RESTful API接口
- WebSocket实时数据推送
- CORS跨域支持
- 请求日志记录
- 并发处理优化
#### 6. Infrastructure 模块(基础设施)
**位置**: `Infrastructure/`
**功能**: 系统级工具和缓存管理
- **Utils/**: 系统工具类(防火墙、图标、网络等)
- **Models/**: 基础设施模型
- **GlobalCaches/**: 全局缓存管理
#### 7. Common 模块(公共工具)
**位置**: `Common/`
**功能**: 通用工具和基础设施
- **Logger/**: 日志系统(基于Serilog)
- **Configuration/**: 配置管理
- **Utils/**: 网络工具、JSON工具等
- **ExpandFun/**: 扩展方法
### 工具类架构
#### 网络监控工具
- **NetworkHelper.cs**: 网络数据格式化和验证
- **NetworkInfoUtil.cs**: 网络接口监控
- **BasicNetworkMonitor.cs**: 基础网络监控
- **EnhancedNetworkMonitor.cs**: 增强网络监控
#### 系统工具
- **SysHelper.cs**: 系统辅助功能
- **SysInfoUtils.cs**: 系统信息获取
- **IconHelper.cs**: 图标提取工具
#### 数据处理工具
- **JsonHelper.cs**: JSON序列化工具
- **StringExpand.cs**: 字符串扩展方法
- **DatabaseManager.cs**: 数据库管理器
#### UI辅助工具
- **TrayIconManager.cs**: 系统托盘管理
- **TaskbarHelper.cs**: 任务栏工具
- **WebView2Helper.cs**: WebView2检测工具
## API接口文档
### HTTP API接口
#### 应用订阅接口
- `POST /api/subscribe-application`: 订阅应用程序网络信息
- `POST /api/subscribe-process`: 订阅进程网络信息
- `POST /api/subscribe-appinfo`: 订阅应用详细信息
- `POST /api/unsubscribe`: 取消订阅
#### 防火墙管理接口
- `GET /api/firewall/rules`: 查询防火墙规则(支持分页和筛选)
- `POST /api/firewall/rules`: 创建防火墙规则
- `PUT /api/firewall/rules`: 更新防火墙规则
- `DELETE /api/firewall/rules`: 删除防火墙规则
- `GET /api/firewall/status`: 获取防火墙状态
- `GET /api/firewall/statistics`: 获取防火墙统计信息
- `POST /api/firewall/switch`: 防火墙开关控制
#### 网络统计接口
- `GET /api/statistics/interfaces`: 获取网络接口列表
- `GET /api/statistics/available-ranges`: 获取可用时间范围
- `POST /api/statistics/clear-cache`: 清理缓存
- `GET /api/traffic/trends`: 获取流量趋势数据
- `GET /api/traffic/top-apps`: 获取Top应用流量数据
#### 实时数据接口
- `GET /api/network/global/realtime`: 获取全局网络实时数据
- `GET /api/realtime/active-apps`: 获取实时活跃应用
- `GET /api/system/info`: 获取系统信息
- `GET /api/system/collection-stats`: 获取数据收集统计
#### 应用分析接口
- `GET /api/apps/top-traffic`: 获取软件流量排行榜
- `GET /api/apps/network-analysis`: 获取应用详细网络分析
### WebSocket接口
- **连接地址**: `ws://127.0.0.1:{动态端口}`
- **认证方式**: URL参数uuid
- **数据格式**: JSON格式,支持GZIP压缩
- **心跳机制**: ping/pong消息
## 配置文件说明
### appsettings.json
```json
{
"AppSettings": {
"AppName": "NetVizor",
"MaxConnection": 444,
"Version": "1.0.0",
"Logging": {
"EnableConsole": true,
"EnableFile": true,
"LogPath": "logs",
"MaxFileSizeMB": 1,
"RetentionDays": 30,
"CompressDaysThreshold": 3,
"MinimumLevel": "Information"
}
}
}
```
### global.json
```json
{
"sdk": {
"version": "8.0.0",
"rollForward": "latestMajor",
"allowPrerelease": true
}
}
```
## 部署要求
### 系统要求
- **操作系统**: Windows 10/11 (x64)
- **运行时**: .NET 8.0 Runtime
- **WebView2**: Microsoft Edge WebView2 Runtime
- **权限**: 管理员权限(ETW监控需要)
### 依赖组件
- **Microsoft.Diagnostics.Tracing**: ETW事件处理
- **NetFwTypeLib**: Windows防火墙COM接口
- **Serilog**: 日志记录
- **Dapper**: ORM框架
- **Fleck**: WebSocket服务器
## 开发环境配置
我所有的开发环境并非全部在默认位置安装,比如部分环境如 Git 之类的在 D:\Env 下
### 开发工具
- Visual Studio 2022 或 JetBrains Rider
- .NET 8.0 SDK
- Node.js(前端开发)
- SQLite管理工具
### 构建命令
```bash
# 还原包
dotnet restore
# 构建项目
dotnet build
# 发布应用
dotnet publish -c Release --self-contained true -r win-x64
```
## 任务管理
你可以使用TODO.md文档来管理你的任务,或者记录任何开发相关的内容。该文件内容完全由你管理,我不会干预,如果有内容你可以直接清空,它就像你的开发草稿纸。
## 注意事项
1. **管理员权限**: ETW网络监控功能需要管理员权限才能正常工作
2. **防火墙权限**: 防火墙规则管理需要管理员权限
3. **WebView2依赖**: 确保系统已安装WebView2 Runtime
4. **数据库权限**: SQLite数据库文件需要读写权限
5. **端口冲突**: 系统会自动选择可用端口,避免冲突
@@ -32,6 +32,7 @@
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AJsonConverter_00601_002Ecs_002Fl_003AC_0021_003FUsers_003Fyllyo_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F315ee97d207b44bebef0c3c849c1e3fa9fb18_003F65_003Fdcbd50cc_003FJsonConverter_00601_002Ecs/@EntryIndexedValue">ForceIncluded</s:String> <s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AJsonConverter_00601_002Ecs_002Fl_003AC_0021_003FUsers_003Fyllyo_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F315ee97d207b44bebef0c3c849c1e3fa9fb18_003F65_003Fdcbd50cc_003FJsonConverter_00601_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AKernelTraceEventParser_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F25b12134b1124ffbab73641db2e54e5b310830_003F24_003Fd19d02f2_003FKernelTraceEventParser_002Ecs/@EntryIndexedValue">ForceIncluded</s:String> <s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AKernelTraceEventParser_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F25b12134b1124ffbab73641db2e54e5b310830_003F24_003Fd19d02f2_003FKernelTraceEventParser_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AKernelTraceEventParser_002Ecs_002Fl_003AC_0021_003FUsers_003Fyllyo_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F25b12134b1124ffbab73641db2e54e5b310830_003F24_003Fd19d02f2_003FKernelTraceEventParser_002Ecs/@EntryIndexedValue">ForceIncluded</s:String> <s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AKernelTraceEventParser_002Ecs_002Fl_003AC_0021_003FUsers_003Fyllyo_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F25b12134b1124ffbab73641db2e54e5b310830_003F24_003Fd19d02f2_003FKernelTraceEventParser_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AKernelTraceEventParser_002Ecs_002Fl_003AC_0021_003FUsers_003Fyllyo_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FSourcesCache_003Fa6f6ce104ccb1c8ee6ef2ade11b7539258e5b98d4ceefbddca8684356d98_003FKernelTraceEventParser_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ALazy_00601_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F211e6f3b24fa438a92f1815153647ce2c8f908_003F74_003F173162e6_003FLazy_00601_002Ecs/@EntryIndexedValue">ForceIncluded</s:String> <s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ALazy_00601_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F211e6f3b24fa438a92f1815153647ce2c8f908_003F74_003F173162e6_003FLazy_00601_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ALogger_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F59d0b425a1244469855c0039f03f864c27e00_003F43_003F8a29b095_003FLogger_002Ecs/@EntryIndexedValue">ForceIncluded</s:String> <s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ALogger_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F59d0b425a1244469855c0039f03f864c27e00_003F43_003F8a29b095_003FLogger_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AMonitor_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F211e6f3b24fa438a92f1815153647ce2c8f908_003Fb5_003F0c446809_003FMonitor_002Ecs/@EntryIndexedValue">ForceIncluded</s:String> <s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AMonitor_002Ecs_002Fl_003AC_0021_003FUsers_003Fcharlatans_003FAppData_003FRoaming_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F211e6f3b24fa438a92f1815153647ce2c8f908_003Fb5_003F0c446809_003FMonitor_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
@@ -7,6 +7,7 @@ using Common.Logger;
using Common.Net.WebSocketConn; using Common.Net.WebSocketConn;
using Common.Utils; using Common.Utils;
using Utils.ETW.Etw; using Utils.ETW.Etw;
using Utils.ETW.Core;
namespace Shell.Services; namespace Shell.Services;
@@ -14,6 +15,7 @@ public class ServerStartupManager
{ {
private readonly ApiRouteManager _apiRouteManager; private readonly ApiRouteManager _apiRouteManager;
private readonly StaticFileServer _staticFileServer; private readonly StaticFileServer _staticFileServer;
private HighPrecisionNetworkMonitor? _networkMonitor;
public ServerStartupManager() public ServerStartupManager()
{ {
@@ -23,24 +25,45 @@ public class ServerStartupManager
public async Task StartAllServicesAsync() public async Task StartAllServicesAsync()
{ {
await InitializeEtwMonitoringAsync(); await InitializeNetworkMonitoringAsync();
StartWebSocketServer(); StartWebSocketServer();
StartApiServer(); StartApiServer();
StartStaticFileServer(); StartStaticFileServer();
} }
private async Task InitializeEtwMonitoringAsync() private async Task InitializeNetworkMonitoringAsync()
{ {
if (!SysHelper.IsAdministrator()) if (!SysHelper.IsAdministrator())
{ {
Console.WriteLine("此程序需要管理员权限才能运行ETW监控!"); Console.WriteLine("此程序需要管理员权限才能运行网络监控!");
Console.WriteLine("请以管理员身份重新运行程序。"); Console.WriteLine("请以管理员身份重新运行程序。");
Console.ReadKey(); Console.ReadKey();
return; return;
} }
var networkManager = new EnhancedEtwNetworkManager(); try
networkManager.StartMonitoring(); {
// 初始化高精度网络监控器(基于TCP连接表,替代ETW方案)
_networkMonitor = new HighPrecisionNetworkMonitor(
updateIntervalMs: 1000, // 1秒更新间隔
enableDetailedStats: true // 启用详细统计(如果系统支持)
);
// 设置事件处理器
_networkMonitor.OnProcessStatsUpdated += OnProcessNetworkStatsUpdated;
_networkMonitor.OnAllProcessStatsUpdated += OnAllProcessStatsUpdated;
_networkMonitor.OnPerformanceStatsUpdated += OnPerformanceStatsUpdated;
// 启动监控
_networkMonitor.Start();
Log.Info("高精度网络监控已启动 - 基于TCP连接表方案,解决ETW突发流量统计不准确问题");
}
catch (Exception ex)
{
Log.Error($"初始化网络监控失败: {ex.Message}");
throw;
}
} }
private void StartWebSocketServer() private void StartWebSocketServer()
@@ -48,12 +71,12 @@ public class ServerStartupManager
int port = SysHelper.GetAvailablePort(); int port = SysHelper.GetAvailablePort();
AppConfig.Instance.WebSocketPort = port; AppConfig.Instance.WebSocketPort = port;
AppConfig.Instance.WebSocketPath = $"ws://127.0.0.1:{port}"; AppConfig.Instance.WebSocketPath = $"ws://127.0.0.1:{port}";
Log.Information4Ctx($"服务启动在端口: {AppConfig.Instance.WebSocketPort}"); Log.Information4Ctx($"服务启动在端口: {AppConfig.Instance.WebSocketPort}");
Log.Information($"服务完整地址: {AppConfig.Instance.WebSocketPath}"); Log.Information($"服务完整地址: {AppConfig.Instance.WebSocketPath}");
WebSocketManager.Instance.Start(AppConfig.Instance.WebSocketPath); WebSocketManager.Instance.Start(AppConfig.Instance.WebSocketPath);
WebSocketManager.Instance.SubscribeConnectionClosed((args => WebSocketManager.Instance.SubscribeConnectionClosed((args =>
{ {
if (args.Uuid != null) if (args.Uuid != null)
@@ -84,7 +107,7 @@ public class ServerStartupManager
int port = 3000; int port = 3000;
AppConfig.Instance.HttpPort = port; AppConfig.Instance.HttpPort = port;
AppConfig.Instance.HttpPath = $"http://127.0.0.1:{port}"; AppConfig.Instance.HttpPath = $"http://127.0.0.1:{port}";
// _staticFileServer.Start(AppConfig.Instance.HttpPort); // _staticFileServer.Start(AppConfig.Instance.HttpPort);
} }
@@ -92,5 +115,78 @@ public class ServerStartupManager
{ {
_staticFileServer?.Stop(); _staticFileServer?.Stop();
_apiRouteManager?.Stop(); _apiRouteManager?.Stop();
_networkMonitor?.Stop();
_networkMonitor?.Dispose();
} }
}
#region 网络监控事件处理器
/// <summary>
/// 处理单个进程的网络统计更新
/// </summary>
private void OnProcessNetworkStatsUpdated(HighPrecisionNetworkMonitor.ProcessNetworkStats stats)
{
// 可以在这里将统计数据推送给前端或写入数据库
Log.Debug($"进程网络统计更新: {stats}");
// 示例:通过WebSocket推送给前端
// WebSocketManager.Instance.BroadcastMessage("network_stats", stats);
}
/// <summary>
/// 处理所有活跃进程的网络统计批量更新
/// </summary>
private void OnAllProcessStatsUpdated(List<HighPrecisionNetworkMonitor.ProcessNetworkStats> statsList)
{
Log.Debug($"批量网络统计更新: {statsList.Count} 个活跃进程");
// 示例:可以在这里进行数据聚合或向前端推送汇总信息
var totalSpeed = statsList.Sum(s => s.TotalBytesPerSecond);
Log.Info($"系统总网速: {FormatSpeed(totalSpeed)}");
}
/// <summary>
/// 处理监控性能统计更新
/// </summary>
private void OnPerformanceStatsUpdated(HighPrecisionNetworkMonitor.MonitoringPerformanceStats perfStats)
{
// 监控系统本身的性能,用于调试和优化
if (perfStats.LastProcessingTimeMs > 100) // 如果处理时间超过100ms则记录警告
{
Log.Warning($"网络监控处理耗时较长: {perfStats}");
}
}
/// <summary>
/// 格式化网速显示
/// </summary>
private static string FormatSpeed(double bytesPerSec)
{
if (bytesPerSec >= 1024 * 1024 * 1024)
return $"{bytesPerSec / (1024.0 * 1024 * 1024):F1} GB/s";
if (bytesPerSec >= 1024 * 1024)
return $"{bytesPerSec / (1024.0 * 1024):F1} MB/s";
if (bytesPerSec >= 1024)
return $"{bytesPerSec / 1024.0:F1} KB/s";
return $"{bytesPerSec:F0} B/s";
}
/// <summary>
/// 获取指定进程的网络统计(供外部调用)
/// </summary>
public HighPrecisionNetworkMonitor.ProcessNetworkStats? GetProcessNetworkStats(uint processId)
{
return _networkMonitor?.GetProcessStats(processId);
}
/// <summary>
/// 获取所有活跃进程的网络统计(供外部调用)
/// </summary>
public List<HighPrecisionNetworkMonitor.ProcessNetworkStats> GetAllActiveProcessStats()
{
return _networkMonitor?.GetAllActiveProcessStats() ??
new List<HighPrecisionNetworkMonitor.ProcessNetworkStats>();
}
#endregion
}
+41 -149
View File
@@ -1,160 +1,52 @@
# NetVizor Frontend - API 定义文档 ## 🚀 NetVizor v1.0.0 — 初版发布
## 1. 网络接口API **NetVizor** 是一款基于 Windows 的现代化网络监控与防火墙管理工具,开源、免费、可视化体验极佳。
它能帮你实时查看电脑上每个软件的网络连接详情、流量使用情况、防火墙规则,并支持历史数据分析。
### 1.1 获取网络接口列表(在全局网速数据表中,把存在的网卡都展示出来,其中需要传递timeRange,以表示拿哪里的数据,day指的是1天的数据,你应该去 GlobalNetworkHourly 中寻找,因为其内部最小单位为小时。如果是 week 则应该去 GlobalNetworkDaily 因为这个表的时间为天,mouth 也对应的是GlobalNetworkDaily,如果是 hour 则应该去 GlobalNetwork 找,他内部的时间是 5 秒。这个值一共4中可能,hour,day,week,mouth,默认 hour 和传统的 SimpleWall、TinyWall 等相比,NetVizor 拥有更直观的 UI、更全面的数据分析功能,以及现代化的交互体验。
目前数据库存储的有对应的 网卡ID,需要查询到所有的网卡ID再单独获取对应的网卡信息并返回 Common\Utils\NetworkAdapterHelper.cs 提供了网卡信息的获取方法
```
GET /api/statistics/interfaces?timeRange=day
```
**返回:**
```json
{
"data": [
{
"id": "eth0",
"name": "以太网",
"displayName": "Realtek PCIe GbE Family Controller",
"isActive": true,
"macAddress": "00:11:22:33:44:55"
},
{
"id": "wifi0",
"name": "WiFi",
"displayName": "Intel Wi-Fi 6 AX200",
"isActive": false,
"macAddress": "66:77:88:99:AA:BB"
}
]
}
```
### 1.2 获取可用时间范围(这里的逻辑是是否有对应的数据,因为我们的数据库会定时的收集和整理信息,如果用户刚安装软件,肯定不会有月数据,也不会有周数据,所以需要据此来确定用户能展示的数据范围,同样分为 hour,day,week,mouth 返回这四个等级就行,hour只要用户安装超过10秒就有数据了,所以最低返回 hour 就行 ---
```
GET /api/statistics/available-ranges
```
**返回:**
```json
{
"data": {
"type": "hour",
"name": "1小时",
"available": true,
"startTime": "2025-01-08T15:00:00Z"
}
}
```
## 2. 流量数据API ### 快速上手
### 2.1 获取流量趋势数据 ### ✨ 核心功能
```
GET /api/traffic/trends
```
**参数:**
- `timeRange`: 时间范围 (1hour|1day|7days|30days)
- `interfaceId`: 网卡ID,可选,默认为"all"表示所有网卡
**返回:** #### 1. 悬浮窗 & 系统托盘
```json
{
"data": {
"interface": "all",
"timeRange": "1hour",
"points": [
{
"timestamp": "1754565364",
"uploadSpeed": 1024000,
"downloadSpeed": 5120000
},{
"timestamp": "1754565364",
"uploadSpeed": 1024000,
"downloadSpeed": 5120000
},{
"timestamp": "1754565364",
"uploadSpeed": 1024000,
"downloadSpeed": 5120000
},{
"timestamp": "1754565364",
"uploadSpeed": 1024000,
"downloadSpeed": 5120000
}
]
}
}
```
### 2.2 获取Top应用流量数据 * 类似 TrafficMonitor 的悬浮窗,实时显示总上传/下载速度。
``` * 支持自定义背景色、文字色、布局方向、网速单位等。
GET /api/traffic/top-apps * 可选 **网速排行榜**,实时显示最占带宽的应用。
```
**参数:**
- `timeRange`: 时间范围
- `limit`: 返回数量,默认10
**返回:** #### 2. 实时监控
```json
{
"data": [
{
"processName": "chrome.exe",
"displayName": "Google Chrome",
"icon": "+Uj0Iu09gH8tBUABUAB2HeA3IEEdYdHLY0EAAAAASUVORK5CYII=",
"totalBytes": 1073741824
},{
"processName": "chrome.exe",
"displayName": "Google Chrome",
"icon": "+Uj0Iu09gH8tBUABUAB2HeA3IEEdYdHLY0EAAAAASUVORK5CYII=",
"totalBytes": 1073741824
},{
"processName": "chrome.exe",
"displayName": "Google Chrome",
"icon": "+Uj0Iu09gH8tBUABUAB2HeA3IEEdYdHLY0EAAAAASUVORK5CYII=",
"totalBytes": 1073741824
}
]
}
```
## 3. 软件排行榜API * **软件列表**:显示所有联网软件的图标、名称、线程数、内存占用等。
* **详细信息**:查看实时网速、PID、子线程信息、连接时长等。
* **连接详情**:协议、状态、本地/远程 IP 及端口、流量占用、存活时间等。
* **软件元数据**:版本、路径、公司、版权信息。
### 3.1 获取软件流量TOP100排行 #### 3. 防火墙管理
```
GET /api/apps/top-traffic
```
**参数:**
- `timeRange`: 时间范围 (1hour|1day|7days|30days)
- `page`: 页码,默认1
- `pageSize`: 每页数量,默认100
**返回:** * 可视化查看 **启用/禁用规则、入站/出站规则**。
```json * 支持新建、编辑、删除、开启/关闭规则。
{ * 列表支持多条件筛选 & 即时搜索,虚拟列表渲染无卡顿。
"data": { * 界面基于 NaiveUI,现代化体验。
"total": 156,
"page": 1, #### 4. 数据分析
"pageSize": 100,
"items": [ * 自动收集全局及各软件的流量数据(每分钟一次)。
{ * 支持查看近 1 小时、24 小时、7 天、30 天的数据。
"rank": 1, * **趋势图**:全局网速变化 & 按网卡过滤。
"processName": "chrome.exe", * **Top 应用流量**:玫瑰图展示占带宽最多的软件。
"displayName": "Google Chrome", * **详细分析**:
"processPath": "C:\\Program Files\\Google\\Chrome\\chrome.exe",
"icon": "H8tBUABUAB2HeA3IEEdYdHLY0EAAAAASUVORK5CYII", * 网络拓扑:可视化连接关系。
"totalBytes": 1073741824, * 流量分析:不同协议占比 & 走势。
"uploadBytes": 104857600, * 连接详情:本地/远程连接频率、流量、持续时间。
"connectionCount": 23 * 端口统计:本地端口使用次数与流量。
},{ * 时间线:每小时带宽与数据占用情况。
"rank": 2,
"processName": "chrome.exe1", ---
"displayName": "Google Chrome1",
"processPath": "C:\\Program Files\\Google\\Chrome\\chrome.exe", ### 📜 开源协议
"icon": "H8tBUABUAB2HeA3IEEdYdHLY0EAAAAASUVORK5CYII",
"totalBytes": 1073741824, 本项目基于 **GPL v3.0** 发布,任何人可自由使用、修改、分发,但必须保留相同协议。
"uploadBytes": 104857600,
"connectionCount": 32
}
]
}
}
```
@@ -0,0 +1,420 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Common.Logger;
using Utils.ETW.Core;
namespace Utils.ETW.Core;
/// <summary>
/// 高精度网络监控器 - 基于TCP连接表的精确流量统计
/// 解决ETW在突发高流量下的统计不准确问题
/// 这是替代ETW方案的核心实现
/// </summary>
public class HighPrecisionNetworkMonitor : IDisposable
{
private readonly Timer _monitorTimer;
private readonly ConcurrentDictionary<string, TcpConnectionStatsCollector.TcpConnectionStats>
_previousStats = new();
private readonly ConcurrentDictionary<uint, ProcessNetworkStats> _processStats = new();
private readonly object _lockObject = new();
// 配置参数
private readonly int _updateIntervalMs;
private readonly bool _enableDetailedStats;
private bool _isRunning = false;
// 性能统计
private long _totalConnections = 0;
private long _activeConnections = 0;
private long _monitoringCycles = 0;
private DateTime _lastUpdate = DateTime.Now;
// 事件
public event Action<ProcessNetworkStats>? OnProcessStatsUpdated;
public event Action<List<ProcessNetworkStats>>? OnAllProcessStatsUpdated;
public event Action<MonitoringPerformanceStats>? OnPerformanceStatsUpdated;
public HighPrecisionNetworkMonitor(int updateIntervalMs = 1000, bool enableDetailedStats = false)
{
_updateIntervalMs = updateIntervalMs;
_enableDetailedStats = enableDetailedStats;
_monitorTimer = new Timer(MonitoringTick, null, Timeout.Infinite, Timeout.Infinite);
Log.Info($"高精度网络监控器已创建 - 更新间隔: {updateIntervalMs}ms, 详细统计: {enableDetailedStats}");
}
/// <summary>
/// 启动监控
/// </summary>
public void Start()
{
if (_isRunning) return;
_isRunning = true;
_lastUpdate = DateTime.Now;
// 启动定时器
_monitorTimer.Change(0, _updateIntervalMs);
Log.Info("高精度网络监控已启动");
}
/// <summary>
/// 停止监控
/// </summary>
public void Stop()
{
if (!_isRunning) return;
_isRunning = false;
_monitorTimer.Change(Timeout.Infinite, Timeout.Infinite);
Log.Info("高精度网络监控已停止");
}
/// <summary>
/// 监控主循环
/// </summary>
private void MonitoringTick(object? state)
{
if (!_isRunning) return;
try
{
var startTime = DateTime.Now;
// 1. 获取所有已建立的TCP连接
var connections = TcpConnectionEnumerator.GetTcpConnections(onlyEstablished: true);
_totalConnections = connections.Count;
// 2. 获取统计信息并计算速率
var currentStats = ProcessConnections(connections);
// 3. 按进程汇总统计
var processStats = AggregateStatsByProcess(currentStats);
// 4. 更新进程统计并触发事件
UpdateProcessStats(processStats);
// 5. 更新性能统计
var processingTime = (DateTime.Now - startTime).TotalMilliseconds;
UpdatePerformanceStats(processingTime);
_monitoringCycles++;
_lastUpdate = DateTime.Now;
}
catch (Exception ex)
{
Log.Error($"网络监控周期执行异常: {ex.Message}");
}
}
/// <summary>
/// 处理连接列表,获取统计信息并计算速率
/// </summary>
private List<TcpConnectionStatsCollector.TcpConnectionStats> ProcessConnections(
List<TcpConnectionEnumerator.TcpConnectionInfo> connections)
{
var currentStatsList = new List<TcpConnectionStatsCollector.TcpConnectionStats>();
_activeConnections = 0;
foreach (var connection in connections)
{
var connectionKey = connection.ConnectionKey;
// 尝试获取连接统计(如果支持详细统计)
TcpConnectionStatsCollector.TcpConnectionStats? currentStats = null;
if (_enableDetailedStats)
{
currentStats = TcpConnectionStatsCollector.GetConnectionStats(connection);
}
// 如果详细统计不可用,创建基础统计记录
if (currentStats == null || !currentStats.StatsAvailable)
{
currentStats = new TcpConnectionStatsCollector.TcpConnectionStats
{
ConnectionKey = connectionKey,
ProcessId = connection.ProcessId,
LocalEndpoint = $"{connection.LocalAddress}:{connection.LocalPort}",
RemoteEndpoint = $"{connection.RemoteAddress}:{connection.RemotePort}",
LastUpdate = DateTime.Now,
StatsAvailable = false,
ErrorMessage = "详细统计不可用,使用基础监控"
};
}
// 如果有上一次的统计数据,计算速率
if (_previousStats.TryGetValue(connectionKey, out var previousStats))
{
currentStats = TcpConnectionStatsCollector.CalculateSpeed(previousStats, currentStats);
// 只有当有速率数据时才计为活跃连接
if (currentStats.BytesInPerSecond > 0 || currentStats.BytesOutPerSecond > 0)
{
_activeConnections++;
}
}
// 更新缓存
_previousStats.AddOrUpdate(connectionKey, currentStats, (key, oldValue) => currentStats);
currentStatsList.Add(currentStats);
}
// 清理已断开连接的统计数据
CleanupDisconnectedConnections(connections);
return currentStatsList;
}
/// <summary>
/// 按进程汇总网络统计
/// </summary>
private Dictionary<uint, ProcessNetworkStats> AggregateStatsByProcess(
List<TcpConnectionStatsCollector.TcpConnectionStats> connectionStats)
{
var processStatsDict = new Dictionary<uint, ProcessNetworkStats>();
var groupedByProcess = connectionStats.GroupBy(s => s.ProcessId);
foreach (var group in groupedByProcess)
{
var processId = group.Key;
var connections = group.ToList();
var processStats = new ProcessNetworkStats
{
ProcessId = processId,
ProcessName = GetProcessName(processId),
ConnectionCount = connections.Count,
LastUpdate = DateTime.Now
};
// 汇总所有连接的流量
foreach (var conn in connections)
{
processStats.TotalBytesIn += conn.TotalBytesIn;
processStats.TotalBytesOut += conn.TotalBytesOut;
processStats.BytesInPerSecond += conn.BytesInPerSecond;
processStats.BytesOutPerSecond += conn.BytesOutPerSecond;
if (conn.StatsAvailable)
{
processStats.DetailedStatsAvailable = true;
}
}
processStats.TotalBytesPerSecond = processStats.BytesInPerSecond + processStats.BytesOutPerSecond;
processStatsDict[processId] = processStats;
}
return processStatsDict;
}
/// <summary>
/// 更新进程统计并触发事件
/// </summary>
private void UpdateProcessStats(Dictionary<uint, ProcessNetworkStats> newStats)
{
var updatedProcesses = new List<ProcessNetworkStats>();
foreach (var kvp in newStats)
{
var processId = kvp.Key;
var newProcessStats = kvp.Value;
// 更新进程统计缓存
_processStats.AddOrUpdate(processId, newProcessStats, (key, oldValue) => newProcessStats);
// 只有活跃进程才触发更新事件(有实际流量)
if (newProcessStats.TotalBytesPerSecond > 0)
{
updatedProcesses.Add(newProcessStats);
OnProcessStatsUpdated?.Invoke(newProcessStats);
}
}
// 触发批量更新事件
if (updatedProcesses.Count > 0)
{
OnAllProcessStatsUpdated?.Invoke(updatedProcesses);
}
}
/// <summary>
/// 清理已断开连接的统计数据
/// </summary>
private void CleanupDisconnectedConnections(List<TcpConnectionEnumerator.TcpConnectionInfo> activeConnections)
{
var activeConnectionKeys = activeConnections.Select(c => c.ConnectionKey).ToHashSet();
var keysToRemove = _previousStats.Keys.Where(key => !activeConnectionKeys.Contains(key)).ToList();
foreach (var key in keysToRemove)
{
_previousStats.TryRemove(key, out _);
}
if (keysToRemove.Count > 0)
{
Log.Debug($"清理了 {keysToRemove.Count} 个已断开连接的统计记录");
}
}
/// <summary>
/// 更新性能统计
/// </summary>
private void UpdatePerformanceStats(double processingTimeMs)
{
var perfStats = new MonitoringPerformanceStats
{
TotalConnections = _totalConnections,
ActiveConnections = _activeConnections,
MonitoringCycles = _monitoringCycles,
LastProcessingTimeMs = processingTimeMs,
LastUpdate = _lastUpdate,
UpdateIntervalMs = _updateIntervalMs,
CachedConnectionStats = _previousStats.Count
};
OnPerformanceStatsUpdated?.Invoke(perfStats);
}
/// <summary>
/// 获取进程名称(带缓存)
/// </summary>
private static readonly ConcurrentDictionary<uint, string> ProcessNameCache = new();
private static string GetProcessName(uint processId)
{
return ProcessNameCache.GetOrAdd(processId, pid =>
{
try
{
using var process = Process.GetProcessById((int)pid);
return process.ProcessName;
}
catch
{
return $"PID_{pid}";
}
});
}
/// <summary>
/// 获取指定进程的网络统计
/// </summary>
public ProcessNetworkStats? GetProcessStats(uint processId)
{
return _processStats.GetValueOrDefault(processId);
}
/// <summary>
/// 获取所有活跃进程的网络统计
/// </summary>
public List<ProcessNetworkStats> GetAllActiveProcessStats()
{
return _processStats.Values
.Where(stats => stats.TotalBytesPerSecond > 0)
.OrderByDescending(stats => stats.TotalBytesPerSecond)
.ToList();
}
/// <summary>
/// 获取网络监控性能统计
/// </summary>
public MonitoringPerformanceStats GetPerformanceStats()
{
return new MonitoringPerformanceStats
{
TotalConnections = _totalConnections,
ActiveConnections = _activeConnections,
MonitoringCycles = _monitoringCycles,
LastUpdate = _lastUpdate,
UpdateIntervalMs = _updateIntervalMs,
CachedConnectionStats = _previousStats.Count
};
}
public void Dispose()
{
Stop();
_monitorTimer?.Dispose();
_previousStats.Clear();
_processStats.Clear();
}
#region 数据模型
/// <summary>
/// 进程网络统计信息
/// </summary>
public class ProcessNetworkStats
{
public uint ProcessId { get; set; }
public string ProcessName { get; set; } = "";
public int ConnectionCount { get; set; }
public ulong TotalBytesIn { get; set; }
public ulong TotalBytesOut { get; set; }
public double BytesInPerSecond { get; set; }
public double BytesOutPerSecond { get; set; }
public double TotalBytesPerSecond { get; set; }
public bool DetailedStatsAvailable { get; set; }
public DateTime LastUpdate { get; set; }
public string FormattedDownloadSpeed => FormatSpeed(BytesInPerSecond);
public string FormattedUploadSpeed => FormatSpeed(BytesOutPerSecond);
public string FormattedTotalSpeed => FormatSpeed(TotalBytesPerSecond);
private static string FormatSpeed(double bytesPerSec)
{
if (bytesPerSec >= 1024 * 1024 * 1024)
return $"{bytesPerSec / (1024.0 * 1024 * 1024):F1} GB/s";
if (bytesPerSec >= 1024 * 1024)
return $"{bytesPerSec / (1024.0 * 1024):F1} MB/s";
if (bytesPerSec >= 1024)
return $"{bytesPerSec / 1024.0:F1} KB/s";
return $"{bytesPerSec:F0} B/s";
}
public override string ToString()
{
return $"{ProcessName} (PID:{ProcessId}): {FormattedTotalSpeed} " +
$"[↓{FormattedDownloadSpeed} ↑{FormattedUploadSpeed}] {ConnectionCount}连接";
}
}
/// <summary>
/// 监控性能统计信息
/// </summary>
public class MonitoringPerformanceStats
{
public long TotalConnections { get; set; }
public long ActiveConnections { get; set; }
public long MonitoringCycles { get; set; }
public double LastProcessingTimeMs { get; set; }
public DateTime LastUpdate { get; set; }
public int UpdateIntervalMs { get; set; }
public int CachedConnectionStats { get; set; }
public double EfficiencyPercentage => TotalConnections > 0
? (ActiveConnections * 100.0) / TotalConnections
: 0;
public override string ToString()
{
return $"监控效率: {ActiveConnections}/{TotalConnections} ({EfficiencyPercentage:F1}%), " +
$"处理耗时: {LastProcessingTimeMs:F1}ms, 周期: {MonitoringCycles}";
}
}
#endregion
}
@@ -0,0 +1,325 @@
using Common.Logger;
using Microsoft.Diagnostics.Tracing;
using Microsoft.Diagnostics.Tracing.Parsers;
using Microsoft.Diagnostics.Tracing.Session;
using Utils.ETW.Models;
using System.Collections.Concurrent;
namespace Utils.ETW.Core;
/// <summary>
/// 优化的ETW网络捕获器 - 解决高带宽情况下的事件丢失问题
/// </summary>
public class OptimizedETWNetworkCapture : IDisposable
{
private readonly TraceEventSession _session;
private readonly ETWTraceEventSource _source;
private bool _isCapturing = false;
// 优化:使用批量处理减少单个事件处理开销
private readonly ConcurrentQueue<TcpConnectionEventData> _tcpEventQueue = new();
private readonly ConcurrentQueue<UdpPacketEventData> _udpEventQueue = new();
private readonly Timer _batchProcessTimer;
// 统计信息
private long _totalEventsReceived = 0;
private long _totalEventsProcessed = 0;
private long _eventsDropped = 0;
public event Action<TcpConnectionEventData>? OnTcpConnectionEvent;
public event Action<UdpPacketEventData>? OnUdpPacketEvent;
// 批量事件处理
public event Action<List<TcpConnectionEventData>>? OnBatchTcpEvents;
public event Action<List<UdpPacketEventData>>? OnBatchUdpEvents;
public OptimizedETWNetworkCapture(string sessionName = "OptimizedNetworkETWSession")
{
// 优化1: 使用TraceEventSession的正确构造函数
// 注意:TraceEventSession的缓冲区配置需要在EnableKernelProvider时设置
_session = new TraceEventSession(sessionName);
_source = new ETWTraceEventSource(sessionName, TraceEventSourceType.Session);
// 优化2: 批量处理定时器,每100ms处理一批事件
_batchProcessTimer = new Timer(ProcessBatchEvents, null, 100, 100);
Log.Info($"已创建优化的ETW会话: {sessionName}");
Log.Info("将在启动时配置更大的缓冲区");
}
public void StartCapture()
{
if (_isCapturing) return;
try
{
EnableNetworkProviders();
SetupOptimizedEventHandlers();
_isCapturing = true;
// 在独立线程中处理事件
Task.Run(() =>
{
try
{
_source.Process();
}
catch (Exception ex)
{
Log.Error($"ETW事件处理异常: {ex.Message}");
}
});
Log.Info("优化的ETW网络捕获已启动");
}
catch (Exception ex)
{
Log.Error($"启动优化ETW捕获失败: {ex.Message}");
throw;
}
}
private void EnableNetworkProviders()
{
// 优化3: 启用网络相关的ETW提供程序
// 注意:.NET TraceEvent库的缓冲区配置方式可能不同
_session.EnableKernelProvider(
KernelTraceEventParser.Keywords.NetworkTCPIP
);
Log.Info("已启用NetworkTCPIP提供程序,使用优化配置");
}
private void SetupOptimizedEventHandlers()
{
var parser = new KernelTraceEventParser(_source);
// 优化4: 最小化事件处理器中的工作量
parser.TcpIpSend += data =>
{
Interlocked.Increment(ref _totalEventsReceived);
// 快速创建事件对象并放入队列,不做复杂处理
var tcpEvent = new TcpConnectionEventData
{
Timestamp = data.TimeStamp,
EventType = "TcpSend",
ProcessId = data.ProcessID,
ThreadId = data.ThreadID,
ProcessName = data.ProcessName,
SourceIp = data.saddr,
DestinationIp = data.daddr,
SourcePort = data.sport,
DestinationPort = data.dport,
Protocol = "TCP",
DataLength = data.size,
SequenceNumber = data.seqnum,
Direction = TrafficDirection.Outbound
};
// 如果队列过长,丢弃最旧的事件
if (_tcpEventQueue.Count > 10000)
{
_tcpEventQueue.TryDequeue(out _);
Interlocked.Increment(ref _eventsDropped);
}
_tcpEventQueue.Enqueue(tcpEvent);
};
parser.TcpIpRecv += data =>
{
Interlocked.Increment(ref _totalEventsReceived);
var tcpEvent = new TcpConnectionEventData
{
Timestamp = data.TimeStamp,
EventType = "TcpReceive",
ProcessId = data.ProcessID,
ThreadId = data.ThreadID,
ProcessName = data.ProcessName,
SourceIp = data.saddr,
DestinationIp = data.daddr,
SourcePort = data.sport,
DestinationPort = data.dport,
Protocol = "TCP",
DataLength = data.size,
SequenceNumber = data.seqnum,
Direction = TrafficDirection.Inbound
};
if (_tcpEventQueue.Count > 10000)
{
_tcpEventQueue.TryDequeue(out _);
Interlocked.Increment(ref _eventsDropped);
}
_tcpEventQueue.Enqueue(tcpEvent);
};
// UDP事件处理(类似优化)
parser.UdpIpSend += data =>
{
Interlocked.Increment(ref _totalEventsReceived);
var udpEvent = new UdpPacketEventData
{
Timestamp = data.TimeStamp,
EventType = "UdpSend",
ProcessId = data.ProcessID,
ProcessName = data.ProcessName,
SourceIp = data.saddr,
DestinationIp = data.daddr,
SourcePort = data.sport,
DestinationPort = data.dport,
Protocol = "UDP",
DataLength = data.size,
Direction = TrafficDirection.Outbound
};
if (_udpEventQueue.Count > 5000)
{
_udpEventQueue.TryDequeue(out _);
Interlocked.Increment(ref _eventsDropped);
}
_udpEventQueue.Enqueue(udpEvent);
};
parser.UdpIpRecv += data =>
{
Interlocked.Increment(ref _totalEventsReceived);
var udpEvent = new UdpPacketEventData
{
Timestamp = data.TimeStamp,
EventType = "UdpReceive",
ProcessId = data.ProcessID,
ProcessName = data.ProcessName,
SourceIp = data.saddr,
DestinationIp = data.daddr,
SourcePort = data.sport,
DestinationPort = data.dport,
Protocol = "UDP",
DataLength = data.size,
Direction = TrafficDirection.Inbound
};
if (_udpEventQueue.Count > 5000)
{
_udpEventQueue.TryDequeue(out _);
Interlocked.Increment(ref _eventsDropped);
}
_udpEventQueue.Enqueue(udpEvent);
};
}
/// <summary>
/// 批量处理事件,减少单个事件的处理开销
/// </summary>
private void ProcessBatchEvents(object? state)
{
try
{
// 处理TCP事件批次
var tcpBatch = new List<TcpConnectionEventData>();
var batchSize = Math.Min(_tcpEventQueue.Count, 1000); // 每批最多1000个
for (int i = 0; i < batchSize; i++)
{
if (_tcpEventQueue.TryDequeue(out var tcpEvent))
{
tcpBatch.Add(tcpEvent);
// 同时触发单个事件(向后兼容)
OnTcpConnectionEvent?.Invoke(tcpEvent);
}
}
if (tcpBatch.Count > 0)
{
OnBatchTcpEvents?.Invoke(tcpBatch);
Interlocked.Add(ref _totalEventsProcessed, tcpBatch.Count);
}
// 处理UDP事件批次
var udpBatch = new List<UdpPacketEventData>();
batchSize = Math.Min(_udpEventQueue.Count, 500); // UDP事件相对较少
for (int i = 0; i < batchSize; i++)
{
if (_udpEventQueue.TryDequeue(out var udpEvent))
{
udpBatch.Add(udpEvent);
OnUdpPacketEvent?.Invoke(udpEvent);
}
}
if (udpBatch.Count > 0)
{
OnBatchUdpEvents?.Invoke(udpBatch);
Interlocked.Add(ref _totalEventsProcessed, udpBatch.Count);
}
// 定期输出统计信息
if (_totalEventsReceived % 10000 == 0 && _totalEventsReceived > 0)
{
var dropRate = (_eventsDropped * 100.0) / _totalEventsReceived;
Log.Info($"ETW统计: 接收={_totalEventsReceived}, 处理={_totalEventsProcessed}, " +
$"丢弃={_eventsDropped} ({dropRate:F2}%), TCP队列={_tcpEventQueue.Count}, UDP队列={_udpEventQueue.Count}");
}
}
catch (Exception ex)
{
Log.Error($"批量处理事件异常: {ex.Message}");
}
}
/// <summary>
/// 获取ETW性能统计
/// </summary>
public EtwPerformanceStats GetPerformanceStats()
{
return new EtwPerformanceStats
{
TotalEventsReceived = _totalEventsReceived,
TotalEventsProcessed = _totalEventsProcessed,
EventsDropped = _eventsDropped,
TcpQueueLength = _tcpEventQueue.Count,
UdpQueueLength = _udpEventQueue.Count,
DropRate = _totalEventsReceived > 0 ? (_eventsDropped * 100.0) / _totalEventsReceived : 0
};
}
public void StopCapture()
{
if (!_isCapturing) return;
_isCapturing = false;
_batchProcessTimer?.Dispose();
_source?.Dispose();
_session?.Dispose();
Log.Info($"ETW捕获已停止 - 最终统计: 接收={_totalEventsReceived}, 处理={_totalEventsProcessed}, 丢弃={_eventsDropped}");
}
public void Dispose()
{
StopCapture();
}
}
/// <summary>
/// ETW性能统计信息
/// </summary>
public class EtwPerformanceStats
{
public long TotalEventsReceived { get; set; }
public long TotalEventsProcessed { get; set; }
public long EventsDropped { get; set; }
public int TcpQueueLength { get; set; }
public int UdpQueueLength { get; set; }
public double DropRate { get; set; }
}
@@ -0,0 +1,261 @@
using System;
using System.Collections.Generic;
using System.Runtime.InteropServices;
using Utils.ETW.Native;
using Common.Logger;
namespace Utils.ETW.Core;
/// <summary>
/// TCP连接枚举器 - 用于获取系统中所有TCP连接信息
/// 这是比ETW更精确的网络流量统计方案的核心组件
/// </summary>
public class TcpConnectionEnumerator
{
private const int AF_INET = 2; // IPv4
/// <summary>
/// TCP连接信息数据模型
/// </summary>
public class TcpConnectionInfo
{
public uint ProcessId { get; set; }
public string LocalAddress { get; set; } = "";
public ushort LocalPort { get; set; }
public string RemoteAddress { get; set; } = "";
public ushort RemotePort { get; set; }
public string State { get; set; } = "";
public TcpConnectionAPI.MIB_TCP_STATE RawState { get; set; }
/// <summary>
/// 连接的唯一标识符(用于跟踪同一连接的统计变化)
/// </summary>
public string ConnectionKey => $"{ProcessId}_{LocalAddress}:{LocalPort}_{RemoteAddress}:{RemotePort}";
/// <summary>
/// 是否为已建立的连接(只有已建立的连接才有流量统计)
/// </summary>
public bool IsEstablished => RawState == TcpConnectionAPI.MIB_TCP_STATE.MIB_TCP_STATE_ESTAB;
/// <summary>
/// 获取MIB_TCPROW结构,用于后续的统计信息查询
/// </summary>
public TcpConnectionAPI.MIB_TCPROW GetTcpRow()
{
// 需要将IP地址和端口转换回网络字节序
return new TcpConnectionAPI.MIB_TCPROW
{
dwLocalAddr = IpAddressToUint(LocalAddress),
dwLocalPort = HostToNetworkOrder(LocalPort),
dwRemoteAddr = IpAddressToUint(RemoteAddress),
dwRemotePort = HostToNetworkOrder(RemotePort)
};
}
private static uint IpAddressToUint(string ipAddress)
{
var parts = ipAddress.Split('.');
if (parts.Length != 4) return 0;
return (uint)(byte.Parse(parts[0]) |
(byte.Parse(parts[1]) << 8) |
(byte.Parse(parts[2]) << 16) |
(byte.Parse(parts[3]) << 24));
}
private static uint HostToNetworkOrder(ushort hostPort)
{
return (uint)((hostPort >> 8) | (hostPort << 8));
}
public override string ToString()
{
return $"PID:{ProcessId} {LocalAddress}:{LocalPort} -> {RemoteAddress}:{RemotePort} [{State}]";
}
}
/// <summary>
/// 获取系统中所有TCP连接
/// </summary>
/// <param name="onlyEstablished">是否只返回已建立的连接</param>
/// <returns>TCP连接列表</returns>
public static List<TcpConnectionInfo> GetTcpConnections(bool onlyEstablished = true)
{
var connections = new List<TcpConnectionInfo>();
try
{
// 第一次调用获取所需缓冲区大小
int bufferSize = 0;
int result = TcpConnectionAPI.GetExtendedTcpTable(
IntPtr.Zero,
ref bufferSize,
true, // 排序
AF_INET,
TcpConnectionAPI.TCP_TABLE_CLASS.TCP_TABLE_OWNER_PID_ALL,
0);
if (result != TcpConnectionAPI.ERROR_INSUFFICIENT_BUFFER)
{
Log.Error($"获取TCP表缓冲区大小失败: {result}");
return connections;
}
// 分配缓冲区
IntPtr buffer = Marshal.AllocHGlobal(bufferSize);
try
{
// 第二次调用获取实际数据
result = TcpConnectionAPI.GetExtendedTcpTable(
buffer,
ref bufferSize,
true,
AF_INET,
TcpConnectionAPI.TCP_TABLE_CLASS.TCP_TABLE_OWNER_PID_ALL,
0);
if (result != TcpConnectionAPI.NO_ERROR)
{
Log.Error($"获取TCP表数据失败: {result}");
return connections;
}
// 解析数据
int entriesCount = Marshal.ReadInt32(buffer);
Log.Debug($"发现 {entriesCount} 个TCP连接");
IntPtr currentPtr = buffer + Marshal.SizeOf<uint>(); // 跳过entries count
for (int i = 0; i < entriesCount; i++)
{
var tcpRow = Marshal.PtrToStructure<TcpConnectionAPI.MIB_TCPROW_OWNER_PID>(currentPtr);
var connectionInfo = new TcpConnectionInfo
{
ProcessId = tcpRow.dwOwningPid,
LocalAddress = TcpConnectionAPI.UintToIpAddress(tcpRow.dwLocalAddr),
LocalPort = TcpConnectionAPI.NetworkToHostOrder(tcpRow.dwLocalPort),
RemoteAddress = TcpConnectionAPI.UintToIpAddress(tcpRow.dwRemoteAddr),
RemotePort = TcpConnectionAPI.NetworkToHostOrder(tcpRow.dwRemotePort),
State = TcpConnectionAPI.GetStateName(tcpRow.dwState),
RawState = tcpRow.dwState
};
// 过滤条件
if (onlyEstablished && !connectionInfo.IsEstablished)
{
currentPtr += Marshal.SizeOf<TcpConnectionAPI.MIB_TCPROW_OWNER_PID>();
continue;
}
connections.Add(connectionInfo);
currentPtr += Marshal.SizeOf<TcpConnectionAPI.MIB_TCPROW_OWNER_PID>();
}
Log.Info($"成功获取 {connections.Count} 个TCP连接" +
(onlyEstablished ? "(仅已建立)" : ""));
}
finally
{
Marshal.FreeHGlobal(buffer);
}
}
catch (Exception ex)
{
Log.Error($"枚举TCP连接时发生异常: {ex.Message}");
}
return connections;
}
/// <summary>
/// 按进程ID分组获取TCP连接
/// </summary>
/// <param name="onlyEstablished">是否只返回已建立的连接</param>
/// <returns>按进程ID分组的TCP连接字典</returns>
public static Dictionary<uint, List<TcpConnectionInfo>> GetTcpConnectionsByProcess(bool onlyEstablished = true)
{
var connectionsByProcess = new Dictionary<uint, List<TcpConnectionInfo>>();
var allConnections = GetTcpConnections(onlyEstablished);
foreach (var connection in allConnections)
{
if (!connectionsByProcess.ContainsKey(connection.ProcessId))
{
connectionsByProcess[connection.ProcessId] = new List<TcpConnectionInfo>();
}
connectionsByProcess[connection.ProcessId].Add(connection);
}
Log.Debug($"TCP连接按进程分组: {connectionsByProcess.Count} 个进程有活跃连接");
return connectionsByProcess;
}
/// <summary>
/// 获取指定进程的TCP连接
/// </summary>
/// <param name="processId">进程ID</param>
/// <param name="onlyEstablished">是否只返回已建立的连接</param>
/// <returns>指定进程的TCP连接列表</returns>
public static List<TcpConnectionInfo> GetTcpConnectionsForProcess(uint processId, bool onlyEstablished = true)
{
var allConnections = GetTcpConnections(onlyEstablished);
var processConnections = allConnections.Where(c => c.ProcessId == processId).ToList();
Log.Debug($"进程 {processId} 有 {processConnections.Count} 个TCP连接");
return processConnections;
}
/// <summary>
/// 获取系统网络连接统计概要
/// </summary>
/// <returns>连接统计信息</returns>
public static NetworkConnectionSummary GetConnectionSummary()
{
var allConnections = GetTcpConnections(false); // 获取所有状态的连接
var summary = new NetworkConnectionSummary();
var groupedByState = allConnections.GroupBy(c => c.RawState);
foreach (var group in groupedByState)
{
var state = group.Key;
var count = group.Count();
switch (state)
{
case TcpConnectionAPI.MIB_TCP_STATE.MIB_TCP_STATE_ESTAB:
summary.EstablishedConnections = count;
break;
case TcpConnectionAPI.MIB_TCP_STATE.MIB_TCP_STATE_LISTEN:
summary.ListeningConnections = count;
break;
case TcpConnectionAPI.MIB_TCP_STATE.MIB_TCP_STATE_TIME_WAIT:
summary.TimeWaitConnections = count;
break;
}
}
summary.TotalConnections = allConnections.Count;
summary.ActiveProcesses = allConnections.Select(c => c.ProcessId).Distinct().Count();
Log.Info($"网络连接概要: 总连接={summary.TotalConnections}, " +
$"已建立={summary.EstablishedConnections}, " +
$"监听={summary.ListeningConnections}, " +
$"活跃进程={summary.ActiveProcesses}");
return summary;
}
/// <summary>
/// 网络连接统计概要
/// </summary>
public class NetworkConnectionSummary
{
public int TotalConnections { get; set; }
public int EstablishedConnections { get; set; }
public int ListeningConnections { get; set; }
public int TimeWaitConnections { get; set; }
public int ActiveProcesses { get; set; }
}
}
@@ -0,0 +1,286 @@
using System;
using System.Collections.Generic;
using System.Runtime.InteropServices;
using Utils.ETW.Native;
using Common.Logger;
namespace Utils.ETW.Core;
/// <summary>
/// TCP连接统计信息收集器
/// 负责获取每个TCP连接的详细流量统计数据
/// </summary>
public class TcpConnectionStatsCollector
{
/// <summary>
/// TCP连接的流量统计数据
/// </summary>
public class TcpConnectionStats
{
public string ConnectionKey { get; set; } = "";
public uint ProcessId { get; set; }
public string LocalEndpoint { get; set; } = "";
public string RemoteEndpoint { get; set; } = "";
// 原始累积数据
public ulong TotalBytesIn { get; set; } // 总接收字节数
public ulong TotalBytesOut { get; set; } // 总发送字节数
public ulong TotalSegmentsIn { get; set; } // 总接收段数
public ulong TotalSegmentsOut { get; set; } // 总发送段数
// 时间戳
public DateTime LastUpdate { get; set; }
// 计算出的速率(需要通过两次采样计算)
public double BytesInPerSecond { get; set; } // 接收速率 bytes/s
public double BytesOutPerSecond { get; set; } // 发送速率 bytes/s
// 统计是否可用
public bool StatsAvailable { get; set; }
public string ErrorMessage { get; set; } = "";
public override string ToString()
{
return $"PID:{ProcessId} {LocalEndpoint}->{RemoteEndpoint} " +
$"In:{FormatBytes(TotalBytesIn)} Out:{FormatBytes(TotalBytesOut)} " +
$"Speed:↓{FormatSpeed(BytesInPerSecond)} ↑{FormatSpeed(BytesOutPerSecond)}";
}
private static string FormatBytes(ulong bytes)
{
if (bytes >= 1024 * 1024 * 1024)
return $"{bytes / (1024.0 * 1024 * 1024):F1}GB";
if (bytes >= 1024 * 1024)
return $"{bytes / (1024.0 * 1024):F1}MB";
if (bytes >= 1024)
return $"{bytes / 1024.0:F1}KB";
return $"{bytes}B";
}
private static string FormatSpeed(double bytesPerSec)
{
if (bytesPerSec >= 1024 * 1024 * 1024)
return $"{bytesPerSec / (1024.0 * 1024 * 1024):F1}GB/s";
if (bytesPerSec >= 1024 * 1024)
return $"{bytesPerSec / (1024.0 * 1024):F1}MB/s";
if (bytesPerSec >= 1024)
return $"{bytesPerSec / 1024.0:F1}KB/s";
return $"{bytesPerSec:F0}B/s";
}
}
/// <summary>
/// 尝试启用TCP连接的统计信息收集
/// 注意:这个功能需要管理员权限,且不是所有连接都支持
/// </summary>
/// <param name="connection">TCP连接信息</param>
/// <returns>是否成功启用</returns>
public static bool EnableConnectionStats(TcpConnectionEnumerator.TcpConnectionInfo connection)
{
try
{
var tcpRow = connection.GetTcpRow();
// 创建启用统计收集的设置
var enableStats = new TcpConnectionAPI.TCP_ESTATS_DATA_RW_v0
{
EnableCollection = TcpConnectionAPI.TCP_BOOLEAN_OPTIONAL.TcpBoolOptEnabled
};
IntPtr settingsPtr = Marshal.AllocHGlobal(Marshal.SizeOf<TcpConnectionAPI.TCP_ESTATS_DATA_RW_v0>());
try
{
Marshal.StructureToPtr(enableStats, settingsPtr, false);
int result = TcpConnectionAPI.SetPerTcpConnectionEStats(
ref tcpRow,
TcpConnectionAPI.TCP_ESTATS_TYPE.TcpConnectionEstatsData,
settingsPtr,
0, // RwVersion
(uint)Marshal.SizeOf<TcpConnectionAPI.TCP_ESTATS_DATA_RW_v0>(),
0); // Offset
if (result == TcpConnectionAPI.NO_ERROR)
{
Log.Debug($"成功启用连接统计: {connection.ConnectionKey}");
return true;
}
else
{
Log.Debug($"启用连接统计失败: {connection.ConnectionKey}, 错误代码: {result}");
return false;
}
}
finally
{
Marshal.FreeHGlobal(settingsPtr);
}
}
catch (Exception ex)
{
Log.Warning($"启用连接统计异常: {connection.ConnectionKey}, {ex.Message}");
return false;
}
}
/// <summary>
/// 获取TCP连接的统计信息
/// </summary>
/// <param name="connection">TCP连接信息</param>
/// <returns>统计数据,如果获取失败则返回null</returns>
public static TcpConnectionStats? GetConnectionStats(TcpConnectionEnumerator.TcpConnectionInfo connection)
{
try
{
var tcpRow = connection.GetTcpRow();
var stats = new TcpConnectionStats
{
ConnectionKey = connection.ConnectionKey,
ProcessId = connection.ProcessId,
LocalEndpoint = $"{connection.LocalAddress}:{connection.LocalPort}",
RemoteEndpoint = $"{connection.RemoteAddress}:{connection.RemotePort}",
LastUpdate = DateTime.Now
};
// 分配内存用于接收统计数据
IntPtr dataPtr = Marshal.AllocHGlobal(Marshal.SizeOf<TcpConnectionAPI.TCP_ESTATS_DATA_ROD_v0>());
try
{
int result = TcpConnectionAPI.GetPerTcpConnectionEStats(
ref tcpRow,
TcpConnectionAPI.TCP_ESTATS_TYPE.TcpConnectionEstatsData,
IntPtr.Zero, // Rw - 不需要
0, // RwVersion
0, // RwSize
IntPtr.Zero, // Ros - 不需要
0, // RosVersion
0, // RosSize
dataPtr, // Rod - 我们要的数据
0, // RodVersion
(uint)Marshal.SizeOf<TcpConnectionAPI.TCP_ESTATS_DATA_ROD_v0>()); // RodSize
if (result == TcpConnectionAPI.NO_ERROR)
{
var data = Marshal.PtrToStructure<TcpConnectionAPI.TCP_ESTATS_DATA_ROD_v0>(dataPtr);
stats.TotalBytesIn = data.DataBytesIn;
stats.TotalBytesOut = data.DataBytesOut;
stats.TotalSegmentsIn = data.DataSegsIn;
stats.TotalSegmentsOut = data.DataSegsOut;
stats.StatsAvailable = true;
Log.Debug($"获取连接统计成功: {stats}");
return stats;
}
else
{
// 很多连接可能不支持详细统计,这是正常的
stats.ErrorMessage = $"获取统计失败,错误代码: {result}";
stats.StatsAvailable = false;
Log.Debug($"连接统计不可用: {connection.ConnectionKey}, 错误: {result}");
return stats;
}
}
finally
{
Marshal.FreeHGlobal(dataPtr);
}
}
catch (Exception ex)
{
Log.Warning($"获取连接统计异常: {connection.ConnectionKey}, {ex.Message}");
return null;
}
}
/// <summary>
/// 批量获取多个连接的统计信息
/// </summary>
/// <param name="connections">连接列表</param>
/// <returns>统计信息列表</returns>
public static List<TcpConnectionStats> GetConnectionStatsBatch(
List<TcpConnectionEnumerator.TcpConnectionInfo> connections)
{
var statsList = new List<TcpConnectionStats>();
int successCount = 0;
int failureCount = 0;
foreach (var connection in connections)
{
var stats = GetConnectionStats(connection);
if (stats != null)
{
statsList.Add(stats);
if (stats.StatsAvailable)
successCount++;
else
failureCount++;
}
else
{
failureCount++;
}
}
Log.Info($"批量获取连接统计完成: 总计={connections.Count}, 成功={successCount}, 失败={failureCount}");
return statsList;
}
/// <summary>
/// 尝试为所有连接启用统计收集
/// 注意:这可能会失败很多次,因为不是所有连接都支持详细统计
/// </summary>
/// <param name="connections">连接列表</param>
/// <returns>成功启用统计的连接数量</returns>
public static int EnableStatsForAllConnections(List<TcpConnectionEnumerator.TcpConnectionInfo> connections)
{
int enabledCount = 0;
foreach (var connection in connections)
{
if (EnableConnectionStats(connection))
{
enabledCount++;
}
}
Log.Info($"尝试启用统计收集: 总连接={connections.Count}, 成功启用={enabledCount}");
return enabledCount;
}
/// <summary>
/// 计算两次采样之间的速率
/// </summary>
/// <param name="previousStats">上一次的统计数据</param>
/// <param name="currentStats">当前的统计数据</param>
/// <returns>包含速率信息的当前统计数据</returns>
public static TcpConnectionStats CalculateSpeed(TcpConnectionStats previousStats, TcpConnectionStats currentStats)
{
if (previousStats == null || currentStats == null)
{
return currentStats;
}
var timeDiff = (currentStats.LastUpdate - previousStats.LastUpdate).TotalSeconds;
if (timeDiff <= 0 || timeDiff > 60) // 避免异常的时间差
{
return currentStats;
}
// 计算字节差值
var bytesInDiff = currentStats.TotalBytesIn >= previousStats.TotalBytesIn
? currentStats.TotalBytesIn - previousStats.TotalBytesIn
: 0; // 处理连接重置的情况
var bytesOutDiff = currentStats.TotalBytesOut >= previousStats.TotalBytesOut
? currentStats.TotalBytesOut - previousStats.TotalBytesOut
: 0;
// 计算速率
currentStats.BytesInPerSecond = bytesInDiff / timeDiff;
currentStats.BytesOutPerSecond = bytesOutDiff / timeDiff;
return currentStats;
}
}
@@ -0,0 +1,388 @@
using Common.Logger;
using Utils.ETW.Core;
using Utils.ETW.Models;
using System.Collections.Concurrent;
using System.Diagnostics;
namespace Utils.ETW.Etw;
/// <summary>
/// 高性能ETW网络管理器 - 解决高带宽下的数据统计问题
/// </summary>
public class HighPerformanceEtwNetworkManager : IDisposable
{
private readonly OptimizedETWNetworkCapture _networkCapture;
private bool _isMonitoring = false;
// 使用高性能数据结构存储流量统计
private readonly ConcurrentDictionary<int, ProcessTrafficStats> _processStats = new();
private readonly ConcurrentDictionary<string, ConnectionTrafficStats> _connectionStats = new();
// 定时器用于定期汇总和清理数据
private readonly Timer _aggregationTimer;
private readonly Timer _cleanupTimer;
// 性能监控
private readonly Stopwatch _performanceWatch = Stopwatch.StartNew();
private long _lastStatsTime = 0;
public event Action<ProcessTrafficSnapshot>? OnProcessTrafficUpdate;
public event Action<List<NetworkConnection>>? OnActiveConnectionsUpdate;
public event Action<EtwPerformanceStats>? OnPerformanceStatsUpdate;
public HighPerformanceEtwNetworkManager()
{
_networkCapture = new OptimizedETWNetworkCapture();
// 订阅批量事件处理(性能更好)
_networkCapture.OnBatchTcpEvents += HandleTcpBatch;
_networkCapture.OnBatchUdpEvents += HandleUdpBatch;
// 每500ms汇总一次流量统计
_aggregationTimer = new Timer(AggregateTrafficStats, null, 500, 500);
// 每30秒清理一次过期数据
_cleanupTimer = new Timer(CleanupExpiredData, null, 30000, 30000);
Log.Info("高性能ETW网络管理器已创建");
}
public void StartMonitoring()
{
if (_isMonitoring) return;
try
{
_networkCapture.StartCapture();
_isMonitoring = true;
_lastStatsTime = _performanceWatch.ElapsedMilliseconds;
Log.Info("高性能ETW网络监控已启动");
}
catch (Exception ex)
{
Log.Error($"启动高性能ETW监控失败: {ex.Message}");
throw;
}
}
public void StopMonitoring()
{
if (!_isMonitoring) return;
_isMonitoring = false;
_networkCapture.StopCapture();
Log.Info("高性能ETW网络监控已停止");
}
/// <summary>
/// 批量处理TCP事件 - 大幅提升性能
/// </summary>
private void HandleTcpBatch(List<TcpConnectionEventData> tcpEvents)
{
var processUpdates = new Dictionary<int, long>();
var connectionUpdates = new Dictionary<string, (long sent, long received)>();
// 批量累积统计,减少锁竞争
foreach (var tcpEvent in tcpEvents)
{
// 更新进程统计
if (!processUpdates.ContainsKey(tcpEvent.ProcessId))
processUpdates[tcpEvent.ProcessId] = 0;
if (tcpEvent.EventType == "TcpSend")
processUpdates[tcpEvent.ProcessId] += tcpEvent.DataLength;
else if (tcpEvent.EventType == "TcpReceive")
processUpdates[tcpEvent.ProcessId] += tcpEvent.DataLength;
// 更新连接统计
var connectionKey =
$"{tcpEvent.ProcessId}_{tcpEvent.SourceIp}:{tcpEvent.SourcePort}_{tcpEvent.DestinationIp}:{tcpEvent.DestinationPort}";
if (!connectionUpdates.ContainsKey(connectionKey))
connectionUpdates[connectionKey] = (0, 0);
var current = connectionUpdates[connectionKey];
if (tcpEvent.EventType == "TcpSend")
connectionUpdates[connectionKey] = (current.sent + tcpEvent.DataLength, current.received);
else if (tcpEvent.EventType == "TcpReceive")
connectionUpdates[connectionKey] = (current.sent, current.received + tcpEvent.DataLength);
}
// 批量更新统计数据
UpdateProcessStats(processUpdates);
UpdateConnectionStats(connectionUpdates, tcpEvents);
}
/// <summary>
/// 批量处理UDP事件
/// </summary>
private void HandleUdpBatch(List<UdpPacketEventData> udpEvents)
{
var processUpdates = new Dictionary<int, long>();
foreach (var udpEvent in udpEvents)
{
if (!processUpdates.ContainsKey(udpEvent.ProcessId))
processUpdates[udpEvent.ProcessId] = 0;
processUpdates[udpEvent.ProcessId] += udpEvent.DataLength;
}
UpdateProcessStats(processUpdates);
}
/// <summary>
/// 批量更新进程统计
/// </summary>
private void UpdateProcessStats(Dictionary<int, long> processUpdates)
{
foreach (var update in processUpdates)
{
_processStats.AddOrUpdate(update.Key,
new ProcessTrafficStats
{
ProcessId = update.Key,
TotalBytes = update.Value,
LastUpdate = DateTime.Now
},
(key, existing) =>
{
existing.TotalBytes += update.Value;
existing.LastUpdate = DateTime.Now;
return existing;
});
}
}
/// <summary>
/// 批量更新连接统计
/// </summary>
private void UpdateConnectionStats(Dictionary<string, (long sent, long received)> connectionUpdates,
List<TcpConnectionEventData> events)
{
foreach (var update in connectionUpdates)
{
var sampleEvent = events.FirstOrDefault();
if (sampleEvent != null)
{
_connectionStats.AddOrUpdate(update.Key,
new ConnectionTrafficStats
{
ProcessId = sampleEvent.ProcessId,
SourceIp = sampleEvent.SourceIp?.ToString() ?? "",
SourcePort = sampleEvent.SourcePort,
DestinationIp = sampleEvent.DestinationIp?.ToString() ?? "",
DestinationPort = sampleEvent.DestinationPort,
BytesSent = update.Value.sent,
BytesReceived = update.Value.received,
LastUpdate = DateTime.Now
},
(key, existing) =>
{
existing.BytesSent += update.Value.sent;
existing.BytesReceived += update.Value.received;
existing.LastUpdate = DateTime.Now;
return existing;
});
}
}
}
/// <summary>
/// 定期汇总流量统计并通知订阅者
/// </summary>
private void AggregateTrafficStats(object? state)
{
try
{
var currentTime = _performanceWatch.ElapsedMilliseconds;
var timeDelta = (currentTime - _lastStatsTime) / 1000.0; // 转换为秒
if (timeDelta < 0.1) return; // 避免除零和过于频繁的计算
// 计算进程流量快照
var processSnapshots = new List<ProcessTrafficSnapshot>();
foreach (var processStats in _processStats.Values)
{
processSnapshots.Add(new ProcessTrafficSnapshot
{
ProcessId = processStats.ProcessId,
ProcessName = GetProcessName(processStats.ProcessId),
TotalBytes = processStats.TotalBytes,
BytesPerSecond = (long)(processStats.TotalBytes / timeDelta),
ConnectionCount = _connectionStats.Values.Count(c => c.ProcessId == processStats.ProcessId),
LastUpdate = processStats.LastUpdate
});
}
// 触发进程流量更新事件
foreach (var snapshot in processSnapshots.Take(50)) // 只发送前50个最活跃的进程
{
OnProcessTrafficUpdate?.Invoke(snapshot);
}
// 触发性能统计事件
OnPerformanceStatsUpdate?.Invoke(_networkCapture.GetPerformanceStats());
_lastStatsTime = currentTime;
}
catch (Exception ex)
{
Log.Error($"汇总流量统计异常: {ex.Message}");
}
}
/// <summary>
/// 清理过期数据
/// </summary>
private void CleanupExpiredData(object? state)
{
try
{
var cutoffTime = DateTime.Now.AddMinutes(-5);
var expiredProcesses = new List<int>();
var expiredConnections = new List<string>();
// 清理过期进程统计
foreach (var kvp in _processStats)
{
if (kvp.Value.LastUpdate < cutoffTime)
expiredProcesses.Add(kvp.Key);
}
foreach (var pid in expiredProcesses)
{
_processStats.TryRemove(pid, out _);
}
// 清理过期连接统计
foreach (var kvp in _connectionStats)
{
if (kvp.Value.LastUpdate < cutoffTime)
expiredConnections.Add(kvp.Key);
}
foreach (var connKey in expiredConnections)
{
_connectionStats.TryRemove(connKey, out _);
}
if (expiredProcesses.Count > 0 || expiredConnections.Count > 0)
{
Log.Info($"清理过期数据: 进程={expiredProcesses.Count}, 连接={expiredConnections.Count}");
}
}
catch (Exception ex)
{
Log.Error($"清理过期数据异常: {ex.Message}");
}
}
/// <summary>
/// 获取进程名称(带缓存)
/// </summary>
private static readonly ConcurrentDictionary<int, string> ProcessNameCache = new();
private string GetProcessName(int processId)
{
return ProcessNameCache.GetOrAdd(processId, pid =>
{
try
{
using var process = Process.GetProcessById(pid);
return process.ProcessName;
}
catch
{
return $"PID_{pid}";
}
});
}
/// <summary>
/// 获取当前活跃连接
/// </summary>
public List<NetworkConnection> GetActiveConnections(int topN = 100)
{
return _connectionStats.Values
.Where(c => c.LastUpdate > DateTime.Now.AddMinutes(-2))
.OrderByDescending(c => c.BytesSent + c.BytesReceived)
.Take(topN)
.Select(c => new NetworkConnection
{
ProcessId = c.ProcessId,
ProcessName = GetProcessName(c.ProcessId),
SourceIp = c.SourceIp,
SourcePort = c.SourcePort,
DestinationIp = c.DestinationIp,
DestinationPort = c.DestinationPort,
BytesSent = c.BytesSent,
BytesReceived = c.BytesReceived,
IsActive = true
})
.ToList();
}
/// <summary>
/// 获取ETW性能统计
/// </summary>
public EtwPerformanceStats GetPerformanceStats()
{
return _networkCapture.GetPerformanceStats();
}
public void Dispose()
{
StopMonitoring();
_aggregationTimer?.Dispose();
_cleanupTimer?.Dispose();
_networkCapture?.Dispose();
}
}
#region 数据模型
public class ProcessTrafficStats
{
public int ProcessId { get; set; }
public long TotalBytes { get; set; }
public DateTime LastUpdate { get; set; }
}
public class ConnectionTrafficStats
{
public int ProcessId { get; set; }
public string SourceIp { get; set; } = "";
public int SourcePort { get; set; }
public string DestinationIp { get; set; } = "";
public int DestinationPort { get; set; }
public long BytesSent { get; set; }
public long BytesReceived { get; set; }
public DateTime LastUpdate { get; set; }
}
public class ProcessTrafficSnapshot
{
public int ProcessId { get; set; }
public string ProcessName { get; set; } = "";
public long TotalBytes { get; set; }
public long BytesPerSecond { get; set; }
public int ConnectionCount { get; set; }
public DateTime LastUpdate { get; set; }
}
public class NetworkConnection
{
public int ProcessId { get; set; }
public string ProcessName { get; set; } = "";
public string SourceIp { get; set; } = "";
public int SourcePort { get; set; }
public string DestinationIp { get; set; } = "";
public int DestinationPort { get; set; }
public long BytesSent { get; set; }
public long BytesReceived { get; set; }
public bool IsActive { get; set; }
}
#endregion
@@ -0,0 +1,245 @@
using Common.Logger;
using Utils.ETW.Core;
using Utils.ETW.Etw;
using Utils.ETW.Tools;
namespace Utils.ETW.Examples;
/// <summary>
/// 优化的ETW使用示例 - 解决高带宽下的数据统计问题
/// </summary>
public class OptimizedEtwUsageExample : IDisposable
{
private readonly HighPerformanceEtwNetworkManager _etwManager;
private readonly NetworkMonitoringComparison _comparison;
private readonly Timer _reportTimer;
public OptimizedEtwUsageExample()
{
// 创建高性能ETW管理器
_etwManager = new HighPerformanceEtwNetworkManager();
// 创建网络监控比对工具
_comparison = new NetworkMonitoringComparison();
// 订阅事件
_etwManager.OnProcessTrafficUpdate += OnProcessTrafficUpdate;
_etwManager.OnActiveConnectionsUpdate += OnActiveConnectionsUpdate;
_etwManager.OnPerformanceStatsUpdate += OnPerformanceStatsUpdate;
_comparison.OnComparisonReport += OnComparisonReport;
// 定时输出性能报告
_reportTimer = new Timer(OutputPerformanceReport, null, 10000, 10000);
}
public async Task StartMonitoringAsync()
{
Log.Info("=== 开始优化的ETW网络监控演示 ===");
// 启动ETW监控
_etwManager.StartMonitoring();
// 启动数据比对
_comparison.StartComparison();
Log.Info("监控已启动,请执行网络活动(如Chrome网速测试)来验证性能...");
// 模拟运行30秒
await Task.Delay(30000);
// 输出最终统计报告
OutputFinalReport();
}
private void OnProcessTrafficUpdate(ProcessTrafficSnapshot snapshot)
{
// 只记录高流量进程
if (snapshot.BytesPerSecond > 1024 * 1024) // 1MB/s
{
Log.Info($"[高流量进程] {snapshot.ProcessName} (PID:{snapshot.ProcessId}): " +
$"{FormatBytes(snapshot.BytesPerSecond)}/s, " +
$"连接数={snapshot.ConnectionCount}, " +
$"总流量={FormatBytes(snapshot.TotalBytes)}");
}
}
private void OnActiveConnectionsUpdate(List<NetworkConnection> connections)
{
var topConnections = connections.Take(5);
foreach (var conn in topConnections)
{
Log.Debug($"[活跃连接] {conn.ProcessName}: " +
$"{conn.SourceIp}:{conn.SourcePort} -> {conn.DestinationIp}:{conn.DestinationPort} " +
$"(发送:{FormatBytes(conn.BytesSent)}, 接收:{FormatBytes(conn.BytesReceived)})");
}
}
private void OnPerformanceStatsUpdate(EtwPerformanceStats stats)
{
if (stats.DropRate > 5.0) // 丢包率超过5%时警告
{
Log.Warning($"[ETW性能警告] 事件丢失率过高: {stats.DropRate:F2}%, " +
$"TCP队列长度:{stats.TcpQueueLength}, UDP队列长度:{stats.UdpQueueLength}");
}
}
private void OnComparisonReport(NetworkMonitoringReport report)
{
var global = report.GlobalComparison;
Log.Info($"=== 网络监控数据比对报告 ({report.Timestamp:HH:mm:ss}) ===");
Log.Info($"全局网速: 下载={FormatBytes(global.GlobalDownloadSpeed)}/s, " +
$"上传={FormatBytes(global.GlobalUploadSpeed)}/s, " +
$"总计={FormatBytes(global.GlobalTotalSpeed)}/s");
Log.Info($"ETW统计: 总计={FormatBytes(global.EtwTotalSpeed)}/s");
Log.Info($"数据差异: {FormatBytes(Math.Abs(global.SpeedDifference))}/s " +
$"({Math.Abs(global.SpeedDifferencePercentage):F1}%) - {global.Status}");
if (Math.Abs(global.SpeedDifferencePercentage) > 20)
{
Log.Warning("⚠️ 数据差异较大,可能原因:");
Log.Warning(" 1. ETW事件丢失(缓冲区不足)");
Log.Warning(" 2. 事件处理延迟");
Log.Warning(" 3. 非TCP/IP流量(如VPN、隧道等)");
}
Log.Info($"ETW性能: 接收事件={report.EtwPerformanceStats.TotalEventsReceived}, " +
$"处理事件={report.EtwPerformanceStats.TotalEventsProcessed}, " +
$"丢弃事件={report.EtwPerformanceStats.EventsDropped}, " +
$"丢失率={report.EtwPerformanceStats.DropRate:F2}%");
}
private void OutputPerformanceReport(object? state)
{
try
{
var etwStats = _etwManager.GetPerformanceStats();
var globalComparison = _comparison.GetGlobalComparison();
// 简化的性能报告
if (etwStats.TotalEventsReceived > 0)
{
Log.Info($"[性能报告] ETW事件处理效率: {etwStats.TotalEventsProcessed}/{etwStats.TotalEventsReceived} " +
$"({(etwStats.TotalEventsProcessed * 100.0 / etwStats.TotalEventsReceived):F1}%), " +
$"数据准确度: {(100 - Math.Abs(globalComparison.SpeedDifferencePercentage)):F1}%");
}
}
catch (Exception ex)
{
Log.Error($"输出性能报告异常: {ex.Message}");
}
}
private void OutputFinalReport()
{
Log.Info("=== 最终监控报告 ===");
var etwStats = _etwManager.GetPerformanceStats();
var globalComparison = _comparison.GetGlobalComparison();
var activeConnections = _etwManager.GetActiveConnections(10);
Log.Info($"ETW总体统计:");
Log.Info($" - 总接收事件: {etwStats.TotalEventsReceived:N0}");
Log.Info($" - 总处理事件: {etwStats.TotalEventsProcessed:N0}");
Log.Info($" - 丢弃事件: {etwStats.EventsDropped:N0}");
Log.Info($" - 事件丢失率: {etwStats.DropRate:F2}%");
Log.Info(
$" - 处理效率: {(etwStats.TotalEventsProcessed * 100.0 / Math.Max(etwStats.TotalEventsReceived, 1)):F2}%");
Log.Info($"数据准确性分析:");
Log.Info($" - 最后全局网速: {FormatBytes(globalComparison.GlobalTotalSpeed)}/s");
Log.Info($" - 最后ETW统计: {FormatBytes(globalComparison.EtwTotalSpeed)}/s");
Log.Info($" - 数据差异: {Math.Abs(globalComparison.SpeedDifferencePercentage):F1}%");
Log.Info($" - 准确性评估: {GetAccuracyAssessment(globalComparison.SpeedDifferencePercentage)}");
Log.Info($"活跃连接统计: {activeConnections.Count} 个");
foreach (var conn in activeConnections.Take(5))
{
Log.Info($" - {conn.ProcessName}: {conn.DestinationIp}:{conn.DestinationPort} " +
$"({FormatBytes(conn.BytesSent + conn.BytesReceived)})");
}
// 给出优化建议
if (etwStats.DropRate > 5.0)
{
Log.Warning("🔧 优化建议: ETW事件丢失率过高");
Log.Warning(" 1. 增加ETW缓冲区大小");
Log.Warning(" 2. 提高事件处理线程优先级");
Log.Warning(" 3. 优化事件处理逻辑");
}
if (Math.Abs(globalComparison.SpeedDifferencePercentage) > 30)
{
Log.Warning("🔧 优化建议: 数据准确性需要改进");
Log.Warning(" 1. 检查是否有VPN或隧道流量");
Log.Warning(" 2. 考虑使用WinDivert等内核级监控");
Log.Warning(" 3. 增加对非TCP/UDP协议的支持");
}
}
private static string GetAccuracyAssessment(double differencePercentage)
{
return Math.Abs(differencePercentage) switch
{
< 5 => "优秀 (差异<5%)",
< 15 => "良好 (差异<15%)",
< 30 => "一般 (差异<30%)",
_ => "需要改进 (差异>30%)"
};
}
private static string FormatBytes(long bytes)
{
if (bytes >= 1024 * 1024 * 1024)
return $"{bytes / (1024.0 * 1024 * 1024):F1} GB";
if (bytes >= 1024 * 1024)
return $"{bytes / (1024.0 * 1024):F1} MB";
if (bytes >= 1024)
return $"{bytes / 1024.0:F1} KB";
return $"{bytes} B";
}
public void Dispose()
{
_reportTimer?.Dispose();
_comparison?.Dispose();
_etwManager?.Dispose();
}
}
/// <summary>
/// 快速测试程序入口点
/// </summary>
public class Program
{
public static async Task Main(string[] args)
{
// 检查管理员权限
if (!IsRunAsAdministrator())
{
Console.WriteLine("此程序需要管理员权限来监听ETW事件!");
Console.WriteLine("请以管理员身份重新运行程序。");
Console.ReadKey();
return;
}
Console.WriteLine("优化的ETW网络监控演示程序");
Console.WriteLine("建议: 在另一个窗口中运行Chrome网速测试来验证监控准确性");
Console.WriteLine("按任意键开始监控...");
Console.ReadKey();
using var example = new OptimizedEtwUsageExample();
await example.StartMonitoringAsync();
Console.WriteLine("监控演示完成,按任意键退出...");
Console.ReadKey();
}
private static bool IsRunAsAdministrator()
{
var identity = System.Security.Principal.WindowsIdentity.GetCurrent();
var principal = new System.Security.Principal.WindowsPrincipal(identity);
return principal.IsInRole(System.Security.Principal.WindowsBuiltInRole.Administrator);
}
}
@@ -0,0 +1,232 @@
using System;
using System.Runtime.InteropServices;
namespace Utils.ETW.Native;
/// <summary>
/// TCP连接表相关的Win32 API声明
/// 用于精确统计网络流量,替代ETW方案解决突发流量统计问题
/// </summary>
public static class TcpConnectionAPI
{
#region 常量定义
public const int NO_ERROR = 0;
public const int ERROR_INSUFFICIENT_BUFFER = 122;
public const int ERROR_INVALID_PARAMETER = 87;
// TCP连接状态
public enum MIB_TCP_STATE
{
MIB_TCP_STATE_CLOSED = 1,
MIB_TCP_STATE_LISTEN = 2,
MIB_TCP_STATE_SYN_SENT = 3,
MIB_TCP_STATE_SYN_RCVD = 4,
MIB_TCP_STATE_ESTAB = 5,
MIB_TCP_STATE_FIN_WAIT1 = 6,
MIB_TCP_STATE_FIN_WAIT2 = 7,
MIB_TCP_STATE_CLOSE_WAIT = 8,
MIB_TCP_STATE_CLOSING = 9,
MIB_TCP_STATE_LAST_ACK = 10,
MIB_TCP_STATE_TIME_WAIT = 11,
MIB_TCP_STATE_DELETE_TCB = 12
}
// TCP表类型
public enum TCP_TABLE_CLASS
{
TCP_TABLE_BASIC_LISTENER,
TCP_TABLE_BASIC_CONNECTIONS,
TCP_TABLE_BASIC_ALL,
TCP_TABLE_OWNER_PID_LISTENER,
TCP_TABLE_OWNER_PID_CONNECTIONS,
TCP_TABLE_OWNER_PID_ALL,
TCP_TABLE_OWNER_MODULE_LISTENER,
TCP_TABLE_OWNER_MODULE_CONNECTIONS,
TCP_TABLE_OWNER_MODULE_ALL
}
// 统计信息启用标志
public enum TCP_ESTATS_TYPE
{
TcpConnectionEstatsData,
TcpConnectionEstatsSynOpts,
TcpConnectionEstatsPath,
TcpConnectionEstatsSendBuff,
TcpConnectionEstatsRec,
TcpConnectionEstatsObs,
TcpConnectionEstatsBandwidth,
TcpConnectionEstatsFineRtt,
TcpConnectionEstatsMaximum
}
public enum TCP_BOOLEAN_OPTIONAL
{
TcpBoolOptDisabled = -1,
TcpBoolOptEnabled = 1,
TcpBoolOptUnchanged = -2
}
#endregion
#region 数据结构
/// <summary>
/// TCP连接行(包含进程ID)
/// </summary>
[StructLayout(LayoutKind.Sequential)]
public struct MIB_TCPROW_OWNER_PID
{
public MIB_TCP_STATE dwState;
public uint dwLocalAddr;
public uint dwLocalPort; // 网络字节序
public uint dwRemoteAddr;
public uint dwRemotePort; // 网络字节序
public uint dwOwningPid;
}
/// <summary>
/// TCP连接表(包含进程ID)
/// </summary>
[StructLayout(LayoutKind.Sequential)]
public struct MIB_TCPTABLE_OWNER_PID
{
public uint dwNumEntries;
[MarshalAs(UnmanagedType.ByValArray, SizeConst = 1)]
public MIB_TCPROW_OWNER_PID[] table;
}
/// <summary>
/// TCP连接的4元组标识
/// </summary>
[StructLayout(LayoutKind.Sequential)]
public struct MIB_TCPROW
{
public uint dwLocalAddr;
public uint dwLocalPort;
public uint dwRemoteAddr;
public uint dwRemotePort;
}
/// <summary>
/// TCP连接数据统计信息
/// </summary>
[StructLayout(LayoutKind.Sequential)]
public struct TCP_ESTATS_DATA_ROD_v0
{
public ulong DataBytesOut; // 发送的数据字节数
public ulong DataSegsOut; // 发送的数据段数
public ulong DataBytesIn; // 接收的数据字节数
public ulong DataSegsIn; // 接收的数据段数
public ulong SegsOut; // 发送的总段数
public ulong SegsIn; // 接收的总段数
public uint SoftErrors; // 软错误数
public uint SoftErrorReason; // 软错误原因
public uint SndUna; // 发送未确认序号
public uint SndNxt; // 发送下一个序号
public uint SndMax; // 发送最大序号
public uint ThruBytesAcked; // 确认的吞吐字节数
public uint RcvNxt; // 接收下一个序号
public uint ThruBytesReceived; // 接收的吞吐字节数
}
/// <summary>
/// 统计信息启用设置
/// </summary>
[StructLayout(LayoutKind.Sequential)]
public struct TCP_ESTATS_DATA_RW_v0
{
public TCP_BOOLEAN_OPTIONAL EnableCollection;
}
#endregion
#region Win32 API函数
/// <summary>
/// 获取扩展TCP连接表(包含进程ID)
/// </summary>
[DllImport("iphlpapi.dll", SetLastError = true)]
public static extern int GetExtendedTcpTable(
IntPtr pTcpTable,
ref int dwOutBufLen,
bool bOrder,
int ulAf, // AF_INET = 2
TCP_TABLE_CLASS TableClass,
uint Reserved);
/// <summary>
/// 设置TCP连接的统计信息收集
/// </summary>
[DllImport("iphlpapi.dll", SetLastError = true)]
public static extern int SetPerTcpConnectionEStats(
ref MIB_TCPROW Row,
TCP_ESTATS_TYPE EstatsType,
IntPtr Rw,
uint RwVersion,
uint RwSize,
uint Offset);
/// <summary>
/// 获取TCP连接的统计信息
/// </summary>
[DllImport("iphlpapi.dll", SetLastError = true)]
public static extern int GetPerTcpConnectionEStats(
ref MIB_TCPROW Row,
TCP_ESTATS_TYPE EstatsType,
IntPtr Rw,
uint RwVersion,
uint RwSize,
IntPtr Ros,
uint RosVersion,
uint RosSize,
IntPtr Rod,
uint RodVersion,
uint RodSize);
#endregion
#region 辅助方法
/// <summary>
/// 将网络字节序端口转换为主机字节序
/// </summary>
public static ushort NetworkToHostOrder(uint networkPort)
{
return (ushort)((networkPort & 0xFF) << 8 | (networkPort >> 8) & 0xFF);
}
/// <summary>
/// 将IP地址从uint转换为字符串
/// </summary>
public static string UintToIpAddress(uint ip)
{
return $"{ip & 0xFF}.{(ip >> 8) & 0xFF}.{(ip >> 16) & 0xFF}.{(ip >> 24) & 0xFF}";
}
/// <summary>
/// 获取TCP连接状态的友好名称
/// </summary>
public static string GetStateName(MIB_TCP_STATE state)
{
return state switch
{
MIB_TCP_STATE.MIB_TCP_STATE_CLOSED => "CLOSED",
MIB_TCP_STATE.MIB_TCP_STATE_LISTEN => "LISTEN",
MIB_TCP_STATE.MIB_TCP_STATE_SYN_SENT => "SYN_SENT",
MIB_TCP_STATE.MIB_TCP_STATE_SYN_RCVD => "SYN_RCVD",
MIB_TCP_STATE.MIB_TCP_STATE_ESTAB => "ESTABLISHED",
MIB_TCP_STATE.MIB_TCP_STATE_FIN_WAIT1 => "FIN_WAIT1",
MIB_TCP_STATE.MIB_TCP_STATE_FIN_WAIT2 => "FIN_WAIT2",
MIB_TCP_STATE.MIB_TCP_STATE_CLOSE_WAIT => "CLOSE_WAIT",
MIB_TCP_STATE.MIB_TCP_STATE_CLOSING => "CLOSING",
MIB_TCP_STATE.MIB_TCP_STATE_LAST_ACK => "LAST_ACK",
MIB_TCP_STATE.MIB_TCP_STATE_TIME_WAIT => "TIME_WAIT",
MIB_TCP_STATE.MIB_TCP_STATE_DELETE_TCB => "DELETE_TCB",
_ => "UNKNOWN"
};
}
#endregion
}
@@ -0,0 +1,270 @@
using Common.Logger;
using Common.Utils;
using Utils.ETW.Core;
using Utils.ETW.Etw;
using System.Diagnostics;
namespace Utils.ETW.Tools;
/// <summary>
/// 网络监控数据比对工具 - 用于验证ETW和全局监控的数据一致性
/// </summary>
public class NetworkMonitoringComparison : IDisposable
{
private readonly HighPerformanceEtwNetworkManager _etwManager;
private readonly Timer _comparisonTimer;
private readonly object _lockObject = new();
// 统计数据
private Dictionary<int, ProcessComparison> _processComparisons = new();
private GlobalTrafficComparison _globalComparison = new();
// 配置
private readonly TimeSpan _comparisonInterval = TimeSpan.FromSeconds(5);
private readonly TimeSpan _reportInterval = TimeSpan.FromSeconds(30);
public event Action<NetworkMonitoringReport>? OnComparisonReport;
public NetworkMonitoringComparison()
{
_etwManager = new HighPerformanceEtwNetworkManager();
// 订阅ETW事件
_etwManager.OnProcessTrafficUpdate += OnEtwProcessTrafficUpdate;
// 定期比对数据
_comparisonTimer = new Timer(PerformComparison, null,
_comparisonInterval, _comparisonInterval);
Log.Info("网络监控数据比对工具已初始化");
}
public void StartComparison()
{
_etwManager.StartMonitoring();
Log.Info("网络监控数据比对已开始");
}
public void StopComparison()
{
_etwManager.StopMonitoring();
Log.Info("网络监控数据比对已停止");
}
private void OnEtwProcessTrafficUpdate(ProcessTrafficSnapshot etwSnapshot)
{
lock (_lockObject)
{
if (!_processComparisons.ContainsKey(etwSnapshot.ProcessId))
{
_processComparisons[etwSnapshot.ProcessId] = new ProcessComparison
{
ProcessId = etwSnapshot.ProcessId,
ProcessName = etwSnapshot.ProcessName
};
}
var comparison = _processComparisons[etwSnapshot.ProcessId];
comparison.EtwTotalBytes = etwSnapshot.TotalBytes;
comparison.EtwBytesPerSecond = etwSnapshot.BytesPerSecond;
comparison.EtwConnectionCount = etwSnapshot.ConnectionCount;
comparison.LastEtwUpdate = etwSnapshot.LastUpdate;
}
}
private void PerformComparison(object? state)
{
try
{
// 获取全局网络统计
var globalSpeed = BasicNetworkMonitor.CalculateTotalSpeed();
lock (_lockObject)
{
// 更新全局比对数据
_globalComparison.GlobalDownloadSpeed = (long)globalSpeed.DownloadSpeed;
_globalComparison.GlobalUploadSpeed = (long)globalSpeed.UploadSpeed;
_globalComparison.GlobalTotalSpeed = (long)(globalSpeed.DownloadSpeed + globalSpeed.UploadSpeed);
// 计算ETW总流量
var etwTotalSpeed = _processComparisons.Values.Sum(p => p.EtwBytesPerSecond);
_globalComparison.EtwTotalSpeed = etwTotalSpeed;
// 计算差异
_globalComparison.SpeedDifference =
_globalComparison.GlobalTotalSpeed - _globalComparison.EtwTotalSpeed;
_globalComparison.SpeedDifferencePercentage = _globalComparison.GlobalTotalSpeed > 0
? ((double)_globalComparison.SpeedDifference * 100.0) / _globalComparison.GlobalTotalSpeed
: 0;
_globalComparison.LastUpdate = DateTime.Now;
// 如果差异超过阈值,输出警告
if (Math.Abs(_globalComparison.SpeedDifferencePercentage) > 20) // 20%阈值
{
Log.Warning($"网络监控数据差异过大: " +
$"全局={FormatBytes(_globalComparison.GlobalTotalSpeed)}/s, " +
$"ETW={FormatBytes(_globalComparison.EtwTotalSpeed)}/s, " +
$"差异={_globalComparison.SpeedDifferencePercentage:F1}%");
}
}
// 定期输出详细报告
var now = DateTime.Now;
if (now.Second % 30 == 0) // 每30秒输出一次
{
GenerateAndSendReport();
}
}
catch (Exception ex)
{
Log.Error($"网络监控数据比对异常: {ex.Message}");
}
}
private void GenerateAndSendReport()
{
lock (_lockObject)
{
var report = new NetworkMonitoringReport
{
Timestamp = DateTime.Now,
GlobalComparison = _globalComparison,
ProcessComparisons = _processComparisons.Values.ToList(),
EtwPerformanceStats = _etwManager.GetPerformanceStats(),
ActiveConnections = _etwManager.GetActiveConnections(20)
};
// 计算汇总统计
report.Summary = new ComparisonSummary
{
TotalProcessesMonitored = _processComparisons.Count,
TotalActiveConnections = report.ActiveConnections.Count,
AverageSpeedDifference = _processComparisons.Values
.Where(p => p.GlobalBytesPerSecond > 0)
.Select(p => Math.Abs(p.SpeedDifferencePercentage))
.DefaultIfEmpty(0)
.Average(),
HighDifferenceProcesses = _processComparisons.Values
.Where(p => Math.Abs(p.SpeedDifferencePercentage) > 50)
.Count()
};
OnComparisonReport?.Invoke(report);
// 输出简化日志
Log.Info($"网络监控比对报告: " +
$"全局速度差异={_globalComparison.SpeedDifferencePercentage:F1}%, " +
$"监控进程={report.Summary.TotalProcessesMonitored}, " +
$"活跃连接={report.Summary.TotalActiveConnections}, " +
$"ETW丢包率={report.EtwPerformanceStats.DropRate:F2}%");
}
}
/// <summary>
/// 获取特定进程的详细比对信息
/// </summary>
public ProcessComparison? GetProcessComparison(int processId)
{
lock (_lockObject)
{
return _processComparisons.GetValueOrDefault(processId);
}
}
/// <summary>
/// 获取当前全局比对信息
/// </summary>
public GlobalTrafficComparison GetGlobalComparison()
{
lock (_lockObject)
{
return _globalComparison;
}
}
private static string FormatBytes(long bytes)
{
if (bytes >= 1024 * 1024 * 1024)
return $"{bytes / (1024.0 * 1024 * 1024):F1} GB";
if (bytes >= 1024 * 1024)
return $"{bytes / (1024.0 * 1024):F1} MB";
if (bytes >= 1024)
return $"{bytes / 1024.0:F1} KB";
return $"{bytes} B";
}
public void Dispose()
{
StopComparison();
_comparisonTimer?.Dispose();
_etwManager?.Dispose();
}
}
#region 数据模型
public class ProcessComparison
{
public int ProcessId { get; set; }
public string ProcessName { get; set; } = "";
// ETW统计
public long EtwTotalBytes { get; set; }
public long EtwBytesPerSecond { get; set; }
public int EtwConnectionCount { get; set; }
public DateTime LastEtwUpdate { get; set; }
// 全局网络监控无法直接提供进程级别数据
// 这里主要用于存储推算的数据
public long GlobalBytesPerSecond { get; set; }
// 比对结果
public long SpeedDifference => GlobalBytesPerSecond - EtwBytesPerSecond;
public double SpeedDifferencePercentage => GlobalBytesPerSecond > 0
? (SpeedDifference * 100.0) / GlobalBytesPerSecond
: 0;
}
public class GlobalTrafficComparison
{
public long GlobalDownloadSpeed { get; set; }
public long GlobalUploadSpeed { get; set; }
public long GlobalTotalSpeed { get; set; }
public long EtwTotalSpeed { get; set; }
public long SpeedDifference { get; set; }
public double SpeedDifferencePercentage { get; set; }
public DateTime LastUpdate { get; set; }
public string Status => Math.Abs(SpeedDifferencePercentage) switch
{
< 10 => "正常",
< 30 => "轻微差异",
< 50 => "显著差异",
_ => "严重差异"
};
}
public class NetworkMonitoringReport
{
public DateTime Timestamp { get; set; }
public GlobalTrafficComparison GlobalComparison { get; set; } = new();
public List<ProcessComparison> ProcessComparisons { get; set; } = new();
public EtwPerformanceStats EtwPerformanceStats { get; set; } = new();
public List<NetworkConnection> ActiveConnections { get; set; } = new();
public ComparisonSummary Summary { get; set; } = new();
}
public class ComparisonSummary
{
public int TotalProcessesMonitored { get; set; }
public int TotalActiveConnections { get; set; }
public double AverageSpeedDifference { get; set; }
public int HighDifferenceProcesses { get; set; }
}
#endregion