326 lines
12 KiB
C#
326 lines
12 KiB
C#
using System.Collections.Concurrent;
|
|
using System.Diagnostics;
|
|
using System.Net;
|
|
using System.Net.NetworkInformation;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using Proxd_man.Config;
|
|
using Proxd_man.Config.Outbound;
|
|
using Proxd_man.Subscribe;
|
|
|
|
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");
|
|
|
|
private readonly ILogger<OutboundMonitor> _logger;
|
|
private readonly IOptionsMonitor<ProxdConfig> _configProvider;
|
|
private readonly SubscribeManager _subscribeManager;
|
|
private readonly XrayCoreManager _xrayCoreManager;
|
|
|
|
private ProxdConfig Config => _configProvider.CurrentValue;
|
|
|
|
public OutboundMonitor(ILogger<OutboundMonitor> logger, IOptionsMonitor<ProxdConfig> configProvider,
|
|
SubscribeManager subscribeManager, XrayCoreManager xrayCoreManager)
|
|
{
|
|
_logger = logger;
|
|
_configProvider = configProvider;
|
|
_subscribeManager = subscribeManager;
|
|
_xrayCoreManager = xrayCoreManager;
|
|
subscribeManager.SubscribeUpdated += async subscribeOutbounds =>
|
|
{
|
|
await ReselectOutbound(subscribeOutbounds);
|
|
};
|
|
|
|
_xrayCoreManager.MainXrayCoreProcessStarted += StartMonitor;
|
|
}
|
|
|
|
public async Task ReselectOutbound(IDictionary<SubscribeSource, List<BaseOutboundSetting>> subscribeOutbounds)
|
|
{
|
|
_logger.LogInformation("开始重新选择代理...");
|
|
var outboundLatencies = new ConcurrentDictionary<BaseOutboundSetting, double>();
|
|
foreach (var (source, outboundSettings) in subscribeOutbounds)
|
|
{
|
|
_logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name);
|
|
List<Task> tasks = [];
|
|
var semaphore = new SemaphoreSlim(4);
|
|
foreach (var outboundSetting in outboundSettings)
|
|
{
|
|
var task = Task.Run(async () =>
|
|
{
|
|
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.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);
|
|
|
|
using 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);
|
|
}
|
|
|
|
// 等 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);
|
|
}
|
|
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 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.LogInformation(
|
|
"选择代理 {OutboundSettingName} ({OutboundSettingTag}) 作为新的出站代理, 地址: {OutboundSettingAddress}",
|
|
bestOutbound.Name, bestOutbound.GetTag(), bestOutbound.Address);
|
|
await _xrayCoreManager.StartMainXrayCoreProcess(bestOutbound);
|
|
}
|
|
else
|
|
{
|
|
_logger.LogWarning("没有可用的代理.");
|
|
}
|
|
}
|
|
|
|
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 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(IWebProxy proxy, CancellationToken cancellationToken = default)
|
|
{
|
|
var httpClient = new HttpClient(new HttpClientHandler
|
|
{
|
|
Proxy = proxy,
|
|
UseProxy = true,
|
|
})
|
|
{
|
|
Timeout = TimeSpan.FromSeconds(10)
|
|
};
|
|
|
|
// 发起 8 次请求，取平均值
|
|
var latencies = new long[8];
|
|
for (var i = 0; i < 9; i++)
|
|
{
|
|
// 第一次请求不计入统计, 因为此时代理可能还未完全启动
|
|
if (i == 0)
|
|
{
|
|
await httpClient.GetAsync(HttpLatencyTestUrl, cancellationToken);
|
|
continue;
|
|
}
|
|
|
|
var sw = Stopwatch.StartNew();
|
|
try
|
|
{
|
|
using var response = await httpClient.GetAsync(HttpLatencyTestUrl, cancellationToken);
|
|
sw.Stop();
|
|
response.EnsureSuccessStatusCode();
|
|
latencies[i - 1] = sw.ElapsedMilliseconds;
|
|
}
|
|
catch (Exception)
|
|
{
|
|
sw.Stop();
|
|
latencies[i - 1] = -1;
|
|
}
|
|
|
|
await Task.Delay(400, cancellationToken);
|
|
}
|
|
|
|
return latencies;
|
|
}
|
|
|
|
private static async Task<long[]> MeasureHttpBandwidth(IWebProxy proxy,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var httpClient = new HttpClient(new HttpClientHandler
|
|
{
|
|
Proxy = proxy,
|
|
UseProxy = true,
|
|
})
|
|
{
|
|
Timeout = TimeSpan.FromSeconds(10)
|
|
};
|
|
|
|
// 进行 3 次下载，取平均值
|
|
var bandwidths = new long[3];
|
|
for (var i = 0; i < 3; i++)
|
|
{
|
|
var sw = Stopwatch.StartNew();
|
|
try
|
|
{
|
|
using var response = await httpClient.GetAsync(HttpBandwidthTestUrl, cancellationToken);
|
|
sw.Stop();
|
|
response.EnsureSuccessStatusCode();
|
|
bandwidths[i] = 51200000 / (sw.ElapsedMilliseconds / 1000);
|
|
}
|
|
catch (Exception)
|
|
{
|
|
sw.Stop();
|
|
bandwidths[i] = -1;
|
|
}
|
|
|
|
await Task.Delay(1000, cancellationToken);
|
|
}
|
|
|
|
return bandwidths;
|
|
}
|
|
} |