From 577293620cda43478de7f3591b5111fc5c45dacf Mon Sep 17 00:00:00 2001 From: LamGC Date: Sun, 12 Jan 2025 01:35:57 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E4=BC=98=E5=8C=96=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E6=A0=BC=E5=BC=8F.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Proxd-man/Monitor/OutboundMonitor.cs | 272 +++++++++++++-------------- 1 file changed, 131 insertions(+), 141 deletions(-) diff --git a/Proxd-man/Monitor/OutboundMonitor.cs b/Proxd-man/Monitor/OutboundMonitor.cs index b9730f4..54c5777 100644 --- a/Proxd-man/Monitor/OutboundMonitor.cs +++ b/Proxd-man/Monitor/OutboundMonitor.cs @@ -12,7 +12,6 @@ namespace Proxd_man.Monitor; public class OutboundMonitor { - private static readonly Uri HttpLatencyTestUrl = new("https://cp.cloudflare.com/generate_204"); private static readonly Uri HttpBandwidthTestUrl = new("https://speed.cloudflare.com/__down?bytes=51200000"); @@ -22,8 +21,9 @@ public class OutboundMonitor private readonly XrayCoreManager _xrayCoreManager; private ProxdConfig Config => _configProvider.CurrentValue; - - public OutboundMonitor(ILogger logger, IOptionsMonitor configProvider, SubscribeManager subscribeManager, XrayCoreManager xrayCoreManager) + + public OutboundMonitor(ILogger logger, IOptionsMonitor configProvider, + SubscribeManager subscribeManager, XrayCoreManager xrayCoreManager) { _logger = logger; _configProvider = configProvider; @@ -39,147 +39,135 @@ public class OutboundMonitor public async Task ReselectOutbound(IDictionary> subscribeOutbounds) { - try + _logger.LogInformation("开始重新选择代理..."); + var outboundLatencies = new ConcurrentDictionary(); + foreach (var (source, outboundSettings) in subscribeOutbounds) { - _logger.LogInformation("开始重新选择代理..."); - var outboundLatencies = new ConcurrentDictionary(); - foreach (var (source, outboundSettings) in subscribeOutbounds) + _logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name); + List tasks = []; + var semaphore = new SemaphoreSlim(4); + foreach (var outboundSetting in outboundSettings) { - _logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name); - List tasks = []; - var semaphore = new SemaphoreSlim(4); - foreach (var outboundSetting in outboundSettings) + var task = Task.Run(async () => { - var task = Task.Run(async () => + if (outboundSetting.Address == null) { - if (outboundSetting.Address == null) + _logger.LogDebug("跳过代理 {OutboundSettingName} ({OutboundSettingTag}), 因为 Address 为空", + outboundSetting.Name, outboundSetting.GetTag()); + return; + } + + await semaphore.WaitAsync(); + + try + { + using var ping = new Ping(); + var pingReply = ping.Send(outboundSetting.Address, 5000); + + if (pingReply.Status != IPStatus.Success) { - _logger.LogDebug("跳过代理 {OutboundSettingName} ({OutboundSettingTag}), 因为 Address 为空", - outboundSetting.Name, outboundSetting.GetTag()); + _logger.LogWarning( + "无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, + pingReply.Status); return; } - await semaphore.WaitAsync(); + _logger.LogDebug( + "Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}) 成功, 地址: {OutboundSettingAddress}, 延迟: {PingReply.RoundtripTime}ms", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, + pingReply.RoundtripTime); - try + // 如果 Ping 成功, 则开始测试代理的延迟和带宽 + _logger.LogDebug( + "开始测试代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); + + // 随机选择一个端口进行测试 + var port = (ushort)new Random().Next(50000, 60000); + + using var outboundTestProcess = + await _xrayCoreManager.CreateOutboundTestProcess(port, outboundSetting); + if (!outboundTestProcess.Start()) { - using var ping = new Ping(); - var pingReply = ping.Send(outboundSetting.Address, 5000); + _logger.LogError("无法启动 Xray-Core 进程, 地址: {OutboundSettingAddress}", + outboundSetting.Address); + } + else + { + _logger.LogDebug("Xray-Core 进程已启动, 地址: {OutboundSettingAddress}", + outboundSetting.Address); + _xrayCoreManager.RedirectXrayCoreProcessOutput(outboundTestProcess); + } - if (pingReply.Status != IPStatus.Success) - { - _logger.LogWarning( - "无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}", - outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, - pingReply.Status); - return; - } + // 等 Xray-Core 就绪. + await Task.Delay(200); + _logger.LogDebug("Xray-Core 进程已就绪, 开始测试延迟..."); - _logger.LogDebug( - "Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}) 成功, 地址: {OutboundSettingAddress}, 延迟: {PingReply.RoundtripTime}ms", - outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, - pingReply.RoundtripTime); - - // 如果 Ping 成功, 则开始测试代理的延迟和带宽 - _logger.LogDebug( - "开始测试代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}", + // 先测试代理的延迟. + var (latencies, _) = await MeasureOutbound(port); + if (latencies.Count(l => l == -1) > 4) + { + _logger.LogWarning( + "代理 {OutboundSettingName} ({OutboundSettingTag}) 的延迟测试失败, 地址: {OutboundSettingAddress}", outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); - - // 随机选择一个端口进行测试 - var port = (ushort)new Random().Next(50000, 60000); - var outboundTestProcess = - await _xrayCoreManager.CreateOutboundTestProcess(port, outboundSetting); - - if (!outboundTestProcess.Start()) - { - _logger.LogError("无法启动 Xray-Core 进程, 地址: {OutboundSettingAddress}", - outboundSetting.Address); - } - else - { - _logger.LogDebug("Xray-Core 进程已启动, 地址: {OutboundSettingAddress}", - outboundSetting.Address); - _xrayCoreManager.RedirectXrayCoreProcessOutput(outboundTestProcess); - } - - try - { - // 等 Xray-Core 就绪. - await Task.Delay(200); - _logger.LogDebug("Xray-Core 进程已就绪, 开始测试延迟..."); - - // 先测试代理的延迟. - var (latencies, _) = await MeasureOutbound(port); - if (latencies.Count(l => l == -1) > 4) - { - _logger.LogWarning( - "代理 {OutboundSettingName} ({OutboundSettingTag}) 的延迟测试失败, 地址: {OutboundSettingAddress}", - outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); - return; - } - - // 计算平均延迟. - var avgLatency = latencies.Where(l => l != -1).Average(); - outboundLatencies[outboundSetting] = avgLatency; - _logger.LogDebug( - "代理 {OutboundSettingName} ({OutboundSettingTag}) 的平均延迟为 {AvgLatency}ms, 地址: {OutboundSettingAddress}", - outboundSetting.Name, outboundSetting.GetTag(), avgLatency, - outboundSetting.Address); - } - finally - { - outboundTestProcess.Kill(); - } + return; } - catch (Exception e) - { - _logger.LogError(e, "测试代理 {OutboundSettingName} ({OutboundSettingTag}) 时发生错误, 跳过代理, 地址: {OutboundSettingAddress}", - outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); - } - finally - { - semaphore.Release(); - } - }); - - tasks.Add(task); - } - Task.WaitAll(tasks); - _logger.LogDebug("订阅 {SubscribeName} 的代理测试完成.", source.Name); + // 计算平均延迟. + var avgLatency = latencies.Where(l => l != -1).Average(); + outboundLatencies[outboundSetting] = avgLatency; + _logger.LogDebug( + "代理 {OutboundSettingName} ({OutboundSettingTag}) 的平均延迟为 {AvgLatency}ms, 地址: {OutboundSettingAddress}", + outboundSetting.Name, outboundSetting.GetTag(), avgLatency, + outboundSetting.Address); + } + catch (Exception e) + { + _logger.LogError(e, + "测试代理 {OutboundSettingName} ({OutboundSettingTag}) 时发生错误, 跳过代理, 地址: {OutboundSettingAddress}", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); + } + finally + { + semaphore.Release(); + } + }); + + tasks.Add(task); } - var sortedOutboundSettings = outboundLatencies - .OrderBy(kv => kv.Value) - .ToList(); - - // 展示延迟前10的节点名称和延迟 - var index = 1; - var latenciesRanking = "---------------- 延迟排名 ----------------\n"; - sortedOutboundSettings - .Take(10) - .ToList() - .ForEach(kv => latenciesRanking += $"{index++}. {kv.Key.Name} ({kv.Key.GetTag()}) [{kv.Key.Address}]: {kv.Value}ms\n"); - _logger.LogInformation(latenciesRanking); - - var bestOutbound = sortedOutboundSettings - .FirstOrDefault().Key; - - if (bestOutbound != null) - { - _logger.LogInformation( - "选择代理 {OutboundSettingName} ({OutboundSettingTag}) 作为新的出站代理, 地址: {OutboundSettingAddress}", - bestOutbound.Name, bestOutbound.GetTag(), bestOutbound.Address); - await _xrayCoreManager.StartMainXrayCoreProcess(bestOutbound); - } - else - { - _logger.LogWarning("没有可用的代理."); - } + Task.WaitAll(tasks); + _logger.LogDebug("订阅 {SubscribeName} 的代理测试完成.", source.Name); } - catch (Exception e) + + var sortedOutboundSettings = outboundLatencies + .OrderBy(kv => kv.Value) + .ToList(); + + // 展示延迟前10的节点名称和延迟 + var index = 1; + var latenciesRanking = ""; + sortedOutboundSettings + .Take(10) + .ToList() + .ForEach(kv => + latenciesRanking += $"{index++}. {kv.Key.Name} ({kv.Key.GetTag()}) [{kv.Key.Address}]: {kv.Value}ms\n"); + _logger.LogInformation("---------------- 延迟排名 ----------------\n{latenciesRanking}", latenciesRanking); + + var bestOutbound = sortedOutboundSettings + .FirstOrDefault().Key; + + if (bestOutbound != null) { - _logger.LogError(e, "重新选择代理时发生错误."); + _logger.LogInformation( + "选择代理 {OutboundSettingName} ({OutboundSettingTag}) 作为新的出站代理, 地址: {OutboundSettingAddress}", + bestOutbound.Name, bestOutbound.GetTag(), bestOutbound.Address); + await _xrayCoreManager.StartMainXrayCoreProcess(bestOutbound); + } + else + { + _logger.LogWarning("没有可用的代理."); } } @@ -190,25 +178,25 @@ public class OutboundMonitor { return new WebProxy(new Uri($"http://localhost:{httpInboundSetting.Port}")); } - + var socksInboundSetting = Config.Inbounds.Socks; if (socksInboundSetting != null) { return new WebProxy(new Uri($"socks5://localhost:{socksInboundSetting.Port}")); } - + throw new Exception("无法从配置中获取代理信息."); } private CancellationTokenSource? _monitorTaskCts; - + public async void StartMonitor() { if (_monitorTaskCts != null) { await StopMonitor(); } - + _monitorTaskCts = new CancellationTokenSource(); _ = Task.Run(async () => { @@ -223,14 +211,14 @@ public class OutboundMonitor var webProxyFromConfig = GetWebProxyFromConfig(); var httpLatency = await MeasureHttpLatency(webProxyFromConfig); - + // 检查单次失败率是否在 50% 以上 var failedCount = httpLatency.Count(l => l == -1); if (failedCount >= 6) { _ = ReselectOutbound(_subscribeManager.GetSubscribeOutbounds()); _ = StopMonitor(); - + break; } } @@ -247,11 +235,12 @@ public class OutboundMonitor _monitorTaskCts = null; } } - - private static async Task<(long[] latencies, long[] bandwidths)> MeasureOutbound(ushort port, bool withBandwidth = false, CancellationToken cancellationToken = default) + + private static async Task<(long[] latencies, long[] bandwidths)> MeasureOutbound(ushort port, + bool withBandwidth = false, CancellationToken cancellationToken = default) { var proxy = new WebProxy(new Uri($"http://localhost:{port}")); - + var latencies = await MeasureHttpLatency(proxy, cancellationToken); var bandwidths = withBandwidth ? await MeasureHttpBandwidth(proxy, cancellationToken) : []; return (latencies, bandwidths); @@ -267,7 +256,7 @@ public class OutboundMonitor { Timeout = TimeSpan.FromSeconds(10) }; - + // 发起 8 次请求，取平均值 var latencies = new long[8]; for (var i = 0; i < 9; i++) @@ -278,7 +267,7 @@ public class OutboundMonitor await httpClient.GetAsync(HttpLatencyTestUrl, cancellationToken); continue; } - + var sw = Stopwatch.StartNew(); try { @@ -295,11 +284,12 @@ public class OutboundMonitor await Task.Delay(400, cancellationToken); } - + return latencies; } - - private static async Task MeasureHttpBandwidth(IWebProxy proxy, CancellationToken cancellationToken = default) + + private static async Task MeasureHttpBandwidth(IWebProxy proxy, + CancellationToken cancellationToken = default) { var httpClient = new HttpClient(new HttpClientHandler { @@ -309,7 +299,7 @@ public class OutboundMonitor { Timeout = TimeSpan.FromSeconds(10) }; - + // 进行 3 次下载，取平均值 var bandwidths = new long[3]; for (var i = 0; i < 3; i++) @@ -330,7 +320,7 @@ public class OutboundMonitor await Task.Delay(1000, cancellationToken); } - + return bandwidths; } } \ No newline at end of file