refactor: 优化代码格式.

This commit is contained in:
2025-01-12 01:35:57 +08:00
parent 3f69b516bb
commit 577293620c
+131 -141
View File
@@ -12,7 +12,6 @@ namespace Proxd_man.Monitor;
public class OutboundMonitor public class OutboundMonitor
{ {
private static readonly Uri HttpLatencyTestUrl = new("https://cp.cloudflare.com/generate_204"); 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"); 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 readonly XrayCoreManager _xrayCoreManager;
private ProxdConfig Config => _configProvider.CurrentValue; private ProxdConfig Config => _configProvider.CurrentValue;
public OutboundMonitor(ILogger<OutboundMonitor> logger, IOptionsMonitor<ProxdConfig> configProvider, SubscribeManager subscribeManager, XrayCoreManager xrayCoreManager) public OutboundMonitor(ILogger<OutboundMonitor> logger, IOptionsMonitor<ProxdConfig> configProvider,
SubscribeManager subscribeManager, XrayCoreManager xrayCoreManager)
{ {
_logger = logger; _logger = logger;
_configProvider = configProvider; _configProvider = configProvider;
@@ -39,147 +39,135 @@ public class OutboundMonitor
public async Task ReselectOutbound(IDictionary<SubscribeSource, List<BaseOutboundSetting>> subscribeOutbounds) public async Task ReselectOutbound(IDictionary<SubscribeSource, List<BaseOutboundSetting>> subscribeOutbounds)
{ {
try _logger.LogInformation("开始重新选择代理...");
var outboundLatencies = new ConcurrentDictionary<BaseOutboundSetting, double>();
foreach (var (source, outboundSettings) in subscribeOutbounds)
{ {
_logger.LogInformation("开始重新选择代理..."); _logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name);
var outboundLatencies = new ConcurrentDictionary<BaseOutboundSetting, double>(); List<Task> tasks = [];
foreach (var (source, outboundSettings) in subscribeOutbounds) var semaphore = new SemaphoreSlim(4);
foreach (var outboundSetting in outboundSettings)
{ {
_logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name); var task = Task.Run(async () =>
List<Task> tasks = [];
var semaphore = new SemaphoreSlim(4);
foreach (var outboundSetting in outboundSettings)
{ {
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 为空", _logger.LogWarning(
outboundSetting.Name, outboundSetting.GetTag()); "无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address,
pingReply.Status);
return; 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(); _logger.LogError("无法启动 Xray-Core 进程, 地址: {OutboundSettingAddress}",
var pingReply = ping.Send(outboundSetting.Address, 5000); outboundSetting.Address);
}
else
{
_logger.LogDebug("Xray-Core 进程已启动, 地址: {OutboundSettingAddress}",
outboundSetting.Address);
_xrayCoreManager.RedirectXrayCoreProcessOutput(outboundTestProcess);
}
if (pingReply.Status != IPStatus.Success) // 等 Xray-Core 就绪.
{ await Task.Delay(200);
_logger.LogWarning( _logger.LogDebug("Xray-Core 进程已就绪, 开始测试延迟...");
"无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address,
pingReply.Status);
return;
}
_logger.LogDebug( // 先测试代理的延迟.
"Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}) 成功, 地址: {OutboundSettingAddress}, 延迟: {PingReply.RoundtripTime}ms", var (latencies, _) = await MeasureOutbound(port);
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, if (latencies.Count(l => l == -1) > 4)
pingReply.RoundtripTime); {
_logger.LogWarning(
// 如果 Ping 成功, 则开始测试代理的延迟和带宽 "代理 {OutboundSettingName} ({OutboundSettingTag}) 的延迟测试失败, 地址: {OutboundSettingAddress}",
_logger.LogDebug(
"开始测试代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address);
return;
// 随机选择一个端口进行测试
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();
}
} }
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 Task.WaitAll(tasks);
.OrderBy(kv => kv.Value) _logger.LogDebug("订阅 {SubscribeName} 的代理测试完成.", source.Name);
.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("没有可用的代理.");
}
} }
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}")); return new WebProxy(new Uri($"http://localhost:{httpInboundSetting.Port}"));
} }
var socksInboundSetting = Config.Inbounds.Socks; var socksInboundSetting = Config.Inbounds.Socks;
if (socksInboundSetting != null) if (socksInboundSetting != null)
{ {
return new WebProxy(new Uri($"socks5://localhost:{socksInboundSetting.Port}")); return new WebProxy(new Uri($"socks5://localhost:{socksInboundSetting.Port}"));
} }
throw new Exception("无法从配置中获取代理信息."); throw new Exception("无法从配置中获取代理信息.");
} }
private CancellationTokenSource? _monitorTaskCts; private CancellationTokenSource? _monitorTaskCts;
public async void StartMonitor() public async void StartMonitor()
{ {
if (_monitorTaskCts != null) if (_monitorTaskCts != null)
{ {
await StopMonitor(); await StopMonitor();
} }
_monitorTaskCts = new CancellationTokenSource(); _monitorTaskCts = new CancellationTokenSource();
_ = Task.Run(async () => _ = Task.Run(async () =>
{ {
@@ -223,14 +211,14 @@ public class OutboundMonitor
var webProxyFromConfig = GetWebProxyFromConfig(); var webProxyFromConfig = GetWebProxyFromConfig();
var httpLatency = await MeasureHttpLatency(webProxyFromConfig); var httpLatency = await MeasureHttpLatency(webProxyFromConfig);
// 检查单次失败率是否在 50% 以上 // 检查单次失败率是否在 50% 以上
var failedCount = httpLatency.Count(l => l == -1); var failedCount = httpLatency.Count(l => l == -1);
if (failedCount >= 6) if (failedCount >= 6)
{ {
_ = ReselectOutbound(_subscribeManager.GetSubscribeOutbounds()); _ = ReselectOutbound(_subscribeManager.GetSubscribeOutbounds());
_ = StopMonitor(); _ = StopMonitor();
break; break;
} }
} }
@@ -247,11 +235,12 @@ public class OutboundMonitor
_monitorTaskCts = null; _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 proxy = new WebProxy(new Uri($"http://localhost:{port}"));
var latencies = await MeasureHttpLatency(proxy, cancellationToken); var latencies = await MeasureHttpLatency(proxy, cancellationToken);
var bandwidths = withBandwidth ? await MeasureHttpBandwidth(proxy, cancellationToken) : []; var bandwidths = withBandwidth ? await MeasureHttpBandwidth(proxy, cancellationToken) : [];
return (latencies, bandwidths); return (latencies, bandwidths);
@@ -267,7 +256,7 @@ public class OutboundMonitor
{ {
Timeout = TimeSpan.FromSeconds(10) Timeout = TimeSpan.FromSeconds(10)
}; };
// 发起 8 次请求,取平均值 // 发起 8 次请求,取平均值
var latencies = new long[8]; var latencies = new long[8];
for (var i = 0; i < 9; i++) for (var i = 0; i < 9; i++)
@@ -278,7 +267,7 @@ public class OutboundMonitor
await httpClient.GetAsync(HttpLatencyTestUrl, cancellationToken); await httpClient.GetAsync(HttpLatencyTestUrl, cancellationToken);
continue; continue;
} }
var sw = Stopwatch.StartNew(); var sw = Stopwatch.StartNew();
try try
{ {
@@ -295,11 +284,12 @@ public class OutboundMonitor
await Task.Delay(400, cancellationToken); await Task.Delay(400, cancellationToken);
} }
return latencies; return latencies;
} }
private static async Task<long[]> MeasureHttpBandwidth(IWebProxy proxy, CancellationToken cancellationToken = default) private static async Task<long[]> MeasureHttpBandwidth(IWebProxy proxy,
CancellationToken cancellationToken = default)
{ {
var httpClient = new HttpClient(new HttpClientHandler var httpClient = new HttpClient(new HttpClientHandler
{ {
@@ -309,7 +299,7 @@ public class OutboundMonitor
{ {
Timeout = TimeSpan.FromSeconds(10) Timeout = TimeSpan.FromSeconds(10)
}; };
// 进行 3 次下载,取平均值 // 进行 3 次下载,取平均值
var bandwidths = new long[3]; var bandwidths = new long[3];
for (var i = 0; i < 3; i++) for (var i = 0; i < 3; i++)
@@ -330,7 +320,7 @@ public class OutboundMonitor
await Task.Delay(1000, cancellationToken); await Task.Delay(1000, cancellationToken);
} }
return bandwidths; return bandwidths;
} }
} }