LamGC/proxd-man
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 看起来效果不错.

Browse Source
This commit is contained in:
LamGC
2025-01-11 02:35:39 +08:00
parent 66a0a2de08
commit 9ed913b15d
6 changed files with 372 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Proxd-man/Monitor/OutboundMonitor.cs
+208 -66
View File
@@ -1,4 +1,5 @@
using System.Diagnostics;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Net;
using System.Net.NetworkInformation;
using Microsoft.Extensions.Logging;
@@ -20,7 +21,7 @@ public class OutboundMonitor
private readonly SubscribeManager _subscribeManager;
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)
{
@@ -32,86 +33,230 @@ public class OutboundMonitor
{
await ReselectOutbound(subscribeOutbounds);
};
_xrayCoreManager.MainXrayCoreProcessStarted += StartMonitor;
}
private async Task ReselectOutbound(IDictionary<SubscribeSource, List<BaseOutboundSetting>> subscribeOutbounds)
public async Task ReselectOutbound(IDictionary<SubscribeSource, List<BaseOutboundSetting>> subscribeOutbounds)
{
_logger.LogInformation("开始重新选择代理...");
foreach (var (source, outboundSettings) in subscribeOutbounds)
try
{
_logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name);
foreach (var outboundSetting in outboundSettings)
_logger.LogInformation("开始重新选择代理...");
var outboundLatencies = new ConcurrentDictionary<BaseOutboundSetting, double>();
foreach (var (source, outboundSettings) in subscribeOutbounds)
{
if (outboundSetting.Address == null)
_logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name);
List<Task> tasks = [];
var semaphore = new SemaphoreSlim(4);
foreach (var outboundSetting in outboundSettings)
{
_logger.LogDebug("跳过代理 {OutboundSettingName} ({OutboundSettingTag}), 因为 Address 为空",
outboundSetting.Name, outboundSetting.GetTag());
continue;
}
using var ping = new Ping();
var pingReply = ping.Send(outboundSetting.Address, 5000);
if (pingReply.Status != IPStatus.Success)
{
_logger.LogWarning("无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, pingReply.Status);
continue;
}
// 如果 Ping 成功, 则开始测试代理的延迟和带宽
_logger.LogDebug("开始测试代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address);
// 随机选择一个端口进行测试
var port = (ushort) new Random().Next(50000, 60000);
var outboundTestProcess = _xrayCoreManager.CreateOutboundTestProcess(port, outboundSetting);
outboundTestProcess.Start();
var task = Task.Run(async () =>
{
if (outboundSetting.Address == null)
{
_logger.LogDebug("跳过代理 {OutboundSettingName} ({OutboundSettingTag}), 因为 Address 为空",
outboundSetting.Name, outboundSetting.GetTag());
return;
}
// 等 Xray-Core 就绪.
await Task.Delay(500);
// 先测试代理的延迟.
var (latencies, bandwidths) = await MeasureOutbound(port);
if (latencies.All(l => l == -1))
{
_logger.LogWarning("代理 {OutboundSettingName} ({OutboundSettingTag}) 的延迟测试失败, 地址: {OutboundSettingAddress}",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address);
continue;
await semaphore.WaitAsync();
try
{
using var ping = new Ping();
var pingReply = ping.Send(outboundSetting.Address, 5000);
if (pingReply.Status != IPStatus.Success)
{
_logger.LogWarning(
"无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address,
pingReply.Status);
return;
}
_logger.LogDebug(
"Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}) 成功, 地址: {OutboundSettingAddress}, 延迟: {PingReply.RoundtripTime}ms",
outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address,
pingReply.RoundtripTime);
// 如果 Ping 成功, 则开始测试代理的延迟和带宽
_logger.LogDebug(
"开始测试代理 {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();
}
}
finally
{
semaphore.Release();
}
});
tasks.Add(task);
}
// 计算平均延迟.
var avgLatency = latencies.Where(l => l != -1).Average();
_logger.LogDebug("代理 {OutboundSettingName} ({OutboundSettingTag}) 的平均延迟为 {AvgLatency}ms, 地址: {OutboundSettingAddress}",
outboundSetting.Name, outboundSetting.GetTag(), avgLatency, outboundSetting.Address);
Task.WaitAll(tasks);
_logger.LogDebug("订阅 {SubscribeName} 的代理测试完成.", source.Name);
}
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("没有可用的代理.");
}
}
}
public void StartMonitor()
{
Task.Run(async () =>
catch (Exception e)
{
});
_logger.LogError(e, "重新选择代理时发生错误.");
}
}
private WebProxy GetWebProxyFromConfig()
{
var httpInboundSetting = Config.Inbounds.Http;
if (httpInboundSetting != null)
{
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 () =>
{
while (!_monitorTaskCts.Token.IsCancellationRequested)
{
var mainXrayCoreProcess = _xrayCoreManager.MainXrayCoreProcess;
if (mainXrayCoreProcess == null || mainXrayCoreProcess.HasExited)
{
_logger.LogDebug("因 MainXrayCore 退出, OutboundMonitor 任务结束.");
break;
}
var webProxyFromConfig = GetWebProxyFromConfig();
var httpLatency = await MeasureHttpLatency(webProxyFromConfig);
// 检查单次失败率是否在 50% 以上
var failedCount = httpLatency.Count(l => l == -1);
if (failedCount >= 6)
{
_ = ReselectOutbound(_subscribeManager.GetSubscribeOutbounds());
_ = StopMonitor();
break;
}
}
return Task.CompletedTask;
}, _monitorTaskCts.Token);
}
public async Task StopMonitor()
{
if (_monitorTaskCts != null)
{
await _monitorTaskCts.CancelAsync();
_monitorTaskCts = null;
}
}
private static async Task<(long[] latencies, long[] bandwidths)> MeasureOutbound(ushort port, bool withBandwidth = false, CancellationToken cancellationToken = default)
{
var latencies = await MeasureHttpLatency(port, cancellationToken);
var bandwidths = withBandwidth ? await MeasureHttpBandwidth(port, cancellationToken) : [];
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);
}
private static async Task<long[]> MeasureHttpLatency(ushort port, CancellationToken cancellationToken = default)
private static async Task<long[]> MeasureHttpLatency(IWebProxy proxy, CancellationToken cancellationToken = default)
{
var httpClient = new HttpClient(new HttpClientHandler
{
Proxy = new WebProxy
{
Address = new Uri($"http://127.0.0.1:{port}")
},
Proxy = proxy,
UseProxy = true,
})
{
@@ -149,14 +294,11 @@ public class OutboundMonitor
return latencies;
}
private static async Task<long[]> MeasureHttpBandwidth(ushort port, CancellationToken cancellationToken = default)
private static async Task<long[]> MeasureHttpBandwidth(IWebProxy proxy, CancellationToken cancellationToken = default)
{
var httpClient = new HttpClient(new HttpClientHandler
{
Proxy = new WebProxy
{
Address = new Uri($"http://127.0.0.1:{port}")
},
Proxy = proxy,
UseProxy = true,
})
{