使用 nacos-sdk-csharp 服务订阅机制动态更新Yarp配置的简易Demo

前言

之前查看了一些查看了一些关于 yarp + naocs 的文章，nacos 官方库也提供了 Yrap+Nacos 的扩展，其核心是通过后台任务BackgroundServier 轮询 Nacos 服务列表来动态更新自定义的IProxyConfigProvider,尝试跟着实现了一下，觉得太过复杂与繁琐，官方sdk 本身提供了服务订阅机制，通过Timer轮询Nacos服务列表通知服务更改，具体可查看 ServiceInfoHolder，InstancesChangeNotifier，IEventListener

nacos 环境搭建

这里使用的是r-nacos ，基于rust 编写的更轻量，性能更高的服务 https://github.com/nacos-group/r-nacos
docker-compose.yml

# 集群部署样例,数据目录: ./data
version: '3.8'services:nacos:image: qingpan/rnacos:stablecontainer_name: nacosports:- "8848:8848"- "9848:9848"- "10848:10848"volumes:- ./data:/io:rwenvironment:- RNACOS_INIT_ADMIN_USERNAME=admin- RNACOS_INIT_ADMIN_PASSWORD=admin- RNACOS_HTTP_PORT=8848restart: always

自定义监听者

public class DefaultServiceChangeListner : IEventListener
{/// <summary>/// 服务变更监听器/// </summary>private readonly ILogger _logger;/// <summary>/// 代理配置存储/// </summary>private readonly IProxyConfigStorage _proxyConfigStorage;/// <summary>/// nacos 存储/// </summary>private readonly IServiceProvider _serviceProvider;public DefaultServiceChangeListner(ILogger logger){_logger = logger;}public DefaultServiceChangeListner(ILogger logger, IProxyConfigStorage proxyConfigStorage, IServiceProvider serviceProvider){_logger = logger;_proxyConfigStorage = proxyConfigStorage;_serviceProvider = serviceProvider;}/// <summary>/// 监听事件/// </summary>/// <param name="event"></param>/// <returns></returns>public async Task OnEvent(IEvent @event){if (@event is InstancesChangeEvent e){var service = e.ServiceName;var groupname = e.GroupName;var hosts = e.Hosts;var index = 1;var destinations = new Dictionary<string, DestinationConfig>();foreach (var item in hosts){_logger.LogInformation($"service:{service} groupname:{groupname} host:{item.Ip}:{item.Port} weight:{item.Weight} enabled:{item.Enabled} healthy:{item.Healthy} metadata:{string.Join(",", item.Metadata.Select(m => $"{m.Key}:{m.Value}"))}");if (!item.Healthy){continue;}var metadata = item.Metadata;var tryGetValue = metadata.TryGetValue("ClusterId", out var clusterId);if(!tryGetValue || string.IsNullOrEmpty(clusterId)){_logger.LogWarning($"service:{service} groupname:{groupname} host:{item.Ip}:{item.Port} metadata ClusterId is null or empty");continue;}var key = $"{clusterId}-{index}";DestinationConfig destinationConfig = new DestinationConfig(){Address = $"http://{item.Ip}:{item.Port}"};destinations.Add(key, destinationConfig);index++;}ClusterConfig cluster = new ClusterConfig{ClusterId = "material-zhaojin",Destinations = destinations};_proxyConfigStorage.SetClusterConfig(cluster);_logger.LogInformation($"service:{service} groupname:{groupname} changed");using var scope = _serviceProvider.CreateScope();var proxyConfigChange = scope.ServiceProvider.GetRequiredService<IProxyConfigChange>();proxyConfigChange.Refresh();_logger.LogInformation($"service:{service} groupname:{groupname} refresh success");await Task.CompletedTask;}}

更新配置的相关类

IProxyConfigStorage

public interface IProxyConfigStorage
{/// <summary>/// 群组/// </summary>public string GroupName { get; }/// <summary>/// 服务/// </summary>public string ServiceName { get; }/// <summary>/// 路由配置/// </summary>public RouteConfig RouteConfig { get; }/// <summary>/// 集群配置/// </summary>public ClusterConfig ClusterConfig { get; }/// <summary>///设置路由/// </summary>/// <param name="routeConfig"></param>void SetRouteConfig(RouteConfig routeConfig);/// <summary>/// 设置集群/// </summary>/// <param name="clusterConfig"></param>void SetClusterConfig(ClusterConfig clusterConfig);
}public class ProxyConfigStorage:IProxyConfigStorage
{/// <summary>/// 服务名称/// </summary>public string ServiceName { get; internal set; }/// <summary>/// 群组名称/// </summary>public string GroupName { get;internal set; }/// <summary>/// 路由配置/// </summary>public RouteConfig RouteConfig { get; internal set; }/// <summary>/// 集群配置/// </summary>public ClusterConfig ClusterConfig { get; internal set; }public ProxyConfigStorage(string serviceName,string groupName,RouteConfig routeConfig, ClusterConfig clusterConfig){ServiceName = serviceName ?? throw new ArgumentNullException(nameof(serviceName));GroupName = groupName ?? throw new ArgumentNullException(nameof(groupName));RouteConfig = routeConfig ?? throw new ArgumentNullException(nameof(routeConfig));ClusterConfig = clusterConfig ?? throw new ArgumentNullException(nameof(clusterConfig));}public void SetRouteConfig(RouteConfig routeConfig) => RouteConfig = routeConfig;public void SetClusterConfig(ClusterConfig clusterConfig) => ClusterConfig = clusterConfig;
}

INacosProxyConfigStorage

    /// <summary>/// nacaos 配置存储/// </summary>public interface INacosProxyConfigStorage{/// <summary>/// 加载代理配置/// </summary>(List<RouteConfig> Routes, List<ClusterConfig> Clusters) LoadProxyConfig();/// <summary>/// 加载代理配置/// </summary>/// <returns></returns>Task<(List<RouteConfig> Routes, List<ClusterConfig> Clusters)> LoadProxyConfigAsync();}/// <summary>/// 日志/// </summary>/// <param name="logger"></param>public class NacosProxyConfigStorage(ILogger<INacosProxyConfigStorage> logger, INacosNamingService nacosNamingService,IServiceProvider serviceProvider) : INacosProxyConfigStorage{private ConcurrentDictionary<string, IProxyConfigStorage> proxyConfigStorages = new ConcurrentDictionary<string, IProxyConfigStorage>();public (List<RouteConfig> Routes, List<ClusterConfig> Clusters) LoadProxyConfig(){List<RouteConfig> routes = new List<RouteConfig>();List<ClusterConfig> clusters = new List<ClusterConfig>();if (proxyConfigStorages.Any()){routes = proxyConfigStorages.Values.Select(p => p.RouteConfig).ToList();clusters = proxyConfigStorages.Values.Select(p => p.ClusterConfig).ToList();return (routes, clusters);}logger.LogInformation("同步加载配置");var route = new RouteConfig{RouteId = "material-route",ClusterId = "material-zhaojin",Match = new RouteMatch{Path = "/api/material/{**catch-all}"},Metadata = new Dictionary<string, string>{{ "ExampleMetadata", "Value" }},Timeout = TimeSpan.FromSeconds(30),MaxRequestBodySize = 1024 * 1024 * 1000,};routes.Add(route);var serviceName = "material";var groupName = "zhaojin";IProxyConfigStorage proxyConfigStorage = new ProxyConfigStorage(serviceName,groupName,route,new ClusterConfig());nacosNamingService.Subscribe(proxyConfigStorage.ServiceName, proxyConfigStorage.GroupName, new DefaultServiceChangeListner(logger, proxyConfigStorage, serviceProvider));proxyConfigStorages.AddOrUpdate(proxyConfigStorage.ServiceName, proxyConfigStorage, (key, value) => proxyConfigStorage);return (routes, clusters);}public Task<(List<RouteConfig> Routes, List<ClusterConfig> Clusters)> LoadProxyConfigAsync(){throw new NotImplementedException();}}

IProxyConfigChange

    public interface IProxyConfigChange{/// <summary>/// 刷新/// </summary>void Refresh();/// <summary>/// 异步刷新/// </summary>/// <returns></returns>Task RefreshAsync();}/// <summary>/// This class is responsible for refreshing the proxy configuration when it changes./// </summary>/// <param name="nacosProxyConfigStorage"></param>/// <param name="memoryConfigProvider"></param>public class ProxyConfigChange(INacosProxyConfigStorage nacosProxyConfigStorage,InMemoryConfigProvider memoryConfigProvider) : IProxyConfigChange{public void Refresh(){var proxyConfig = nacosProxyConfigStorage.LoadProxyConfig();memoryConfigProvider.Update(proxyConfig.Routes,proxyConfig.Clusters);}public Task RefreshAsync(){throw new NotImplementedException();}}

服务注册

            builder.Services.AddReverseProxy().AddTransforms(ctx=>{ctx.AddRequestTransform(async transformContext =>{// 1. 获取路由匹配时捕获的 {catch-all} 参数的值// 注意：这里假设您的路由配置中使用了 {**catch-all} 或 {remainder}if (transformContext.HttpContext.Request.RouteValues.TryGetValue("catch-all", out var catchAllValue) ||transformContext.HttpContext.Request.RouteValues.TryGetValue("remainder", out catchAllValue)){var catchAllPath = catchAllValue?.ToString();// 2. 构建新的路径// 将捕获的路径片段拼接到新的基础路径 /api/ 后面var newPath = $"/api/{catchAllPath}";// 3. 安全地更新路径// 使用 PathString 以确保正确处理斜杠transformContext.Path = new PathString(newPath);}// 如果没有捕获到值，transformContext.Path 会保持原样（但根据您的路由，这通常不会发生）await Task.CompletedTask;});});builder.Services.AddTransient<IProxyConfigChange, ProxyConfigChange>();builder.Services.AddSingleton<INacosProxyConfigStorage, NacosProxyConfigStorage>();builder.Services.AddSingleton(sp =>{var proxyConfigStorage = sp.GetRequiredService<INacosProxyConfigStorage>();var proxyConfig = proxyConfigStorage.LoadProxyConfig();return new InMemoryConfigProvider(proxyConfig.Routes, proxyConfig.Clusters);});builder.Services.AddSingleton<IProxyConfigProvider>(sp=>sp.GetRequiredService<InMemoryConfigProvider>());

结果

最后总结

这里只是提供了一个简易的demo，当然本人更倾向于将 yarp 的路由配置持久化到数据库中，集群服务地址由nacos 提供，由于有点繁琐再加上本人有点懒这里就不做实现，只是提供一个简易的demo 供参考，本文中提供的代码过于简陋以及路由配置也是写死了的，好吧，懒到没边了，另外关于yarp 的更多操作可以查看官网Yarp,添加限流，集群节点间的负载均衡策略，以及对于节点健康检查策略