.NET Core 3.0之创建基于Consul的Configuration扩展组件

别总是抱怨生活不够幸运,是你欠生活一份努力。未来美不美,取决于你现在拼不拼!

经过前面三篇关于.NET Core Configuration的文章之后,本篇文章主要讨论如何扩展一个Configuration组件出来。

了解了Configuration的源码后,再去扩展一个组件就会比较简单,接下来我们将在.NET Core 3.0-preview5的基础上创建一个基于Consul的配置组件。

相信大家对Consul已经比较了解了,很多项目都会使用Consul作为配置中心,此处也不做其他阐述了,主要是讲一下,创建Consul配置扩展的一些思路。使用Consul配置功能时,我们可以将信息转成JSON格式后再存储,那么我们在读取的时候,在体验上就像是从读取JSON文件中读取一样。

开发前的准备初始化Consul

假设你已经安装并启动了Consul,我们打开Key/Value功能界面,创建两组配置选项出来,分别是commonservice和userservice,如下图所示

配置值采用JSON格式

实现思路

我们知道在Configuration整个的设计框架里,比较重要的类ConfigurationRoot,内部又有一个IConfigurationProvider集合属性,也就是说我们追加IConfigurationProvider实例最终也会被放到到该集合中,如下图所示

该项目中,我使用到了一个已经封装好的Consul(V0.7.2.6)类库,同时基于.NET Core关于Configuration的设计风格,做如下的框架设计

考虑到我会在该组件内部创建ConsulClient实例,所以对ConsulClient构造函数的一部分参数做了抽象提取,并添加到了IConsulConfigurationSource中,以增强该组件的灵活性。

之前说过,Consul中的配置信息是以JSON格式存储的,所以此处使用到了Microsoft.Extensions.Configuration.Json.JsonConfigurationFileParser,用以将JSON格式的信息转换为Configuration的通用格式Key/Value。

核心代码 IConsulConfigurationSource

 /// <summary>
 /// ConsulConfigurationSource
 /// </summary>
public interface IConsulConfigurationSource : IConfigurationSource
 {
  /// <summary>
  /// CancellationToken
  /// </summary>
  CancellationToken CancellationToken { get; }
 
  /// <summary>
  /// Consul构造函数实例,可自定义传入
  /// </summary>
  Action<ConsulClientConfiguration> ConsulClientConfiguration { get; set; }
 
  /// <summary>
  /// Consul构造函数实例,可自定义传入
  /// </summary>
  Action<HttpClient> ConsulHttpClient { get; set; }
 
  /// <summary>
  /// Consul构造函数实例,可自定义传入
  /// </summary>
  Action<HttpClientHandler> ConsulHttpClientHandler { get; set; }
 
  /// <summary>
  /// 服务名称
  /// </summary>
  string ServiceKey { get; }
 
  /// <summary>
  /// 可选项
  /// </summary>
  bool Optional { get; set; }
 
  /// <summary>
  /// Consul查询选项
  /// </summary>
  QueryOptions QueryOptions { get; set; }
 
  /// <summary>
  /// 重新加载延迟时间,单位是毫秒
  /// </summary>
  int ReloadDelay { get; set; }
 
  /// <summary>
  /// 是否在配置改变的时候重新加载
  /// </summary>
  bool ReloadOnChange { get; set; }
 }

ConsulConfigurationSource

该类提供了一个构造函数,用于接收ServiceKey和CancellationToken实例

 public ConsulConfigurationSource(string serviceKey, CancellationToken cancellationToken)
 {
 if (string.IsNullOrWhiteSpace(serviceKey))
 {
  throw new ArgumentNullException(nameof(serviceKey));
 }
 
 this.ServiceKey = serviceKey;
 this.CancellationToken = cancellationToken;
}

其build()方法也比较简单,主要是初始化ConsulConfigurationParser实例

 public IConfigurationProvider Build(IConfigurationBuilder builder)
 {
  ConsulConfigurationParser consulParser = new ConsulConfigurationParser(this);
 
  return new ConsulConfigurationProvider(this, consulParser);
 }

ConsulConfigurationParser

该类比较复杂,主要实现Consul配置的获取、监控以及容错处理,公共方法源码如下

 /// <summary>
 /// 获取并转换Consul配置信息
 /// </summary>
 /// <param name="reloading"></param>
 /// <param name="source"></param>
 /// <returns></returns>
 public async Task<IDictionary<string, string>> GetConfig(bool reloading, IConsulConfigurationSource source)
 {
  try
  {
   QueryResult<KVPair> kvPair = await this.GetKvPairs(source.ServiceKey, source.QueryOptions, source.CancellationToken).ConfigureAwait(false);
   if ((kvPair?.Response == null) && !source.Optional)
   {
    if (!reloading)
    {
     throw new FormatException(Resources.Error_InvalidService(source.ServiceKey));
    }
 
    return new Dictionary<string, string>();
   }
 
   if (kvPair?.Response == null)
   {
    throw new FormatException(Resources.Error_ValueNotExist(source.ServiceKey));
   }
 
   this.UpdateLastIndex(kvPair);
 
   return JsonConfigurationFileParser.Parse(source.ServiceKey, new MemoryStream(kvPair.Response.Value));
  }
  catch (Exception exception)
  {
   throw exception;
  }
 }
 
 /// <summary>
 /// Consul配置信息监控
 /// </summary>
 /// <param name="key"></param>
 /// <param name="cancellationToken"></param>
 /// <returns></returns>
 public IChangeToken Watch(string key, CancellationToken cancellationToken)
 {
  Task.Run(() => this.RefreshForChanges(key, cancellationToken), cancellationToken);
 
  return this.reloadToken;
 }

另外,关于Consul的监控主要利用了QueryResult.LastIndex属性,该类缓存了该属性的值,并与实获取的值进行比较,以判断是否需要重新加载内存中的缓存配置

ConsulConfigurationProvider

该类除了实现Load方法外,还会根据ReloadOnChange属性,在构造函数中注册OnChange事件,用于重新加载配置信息,源码如下:

public sealed class ConsulConfigurationProvider : ConfigurationProvider
 {
  private readonly ConsulConfigurationParser configurationParser;
  private readonly IConsulConfigurationSource source;
 
  public ConsulConfigurationProvider(IConsulConfigurationSource source, ConsulConfigurationParser configurationParser)
  {
   this.configurationParser = configurationParser;
   this.source = source;
 
   if (source.ReloadOnChange)
   {
    ChangeToken.OnChange(
     () => this.configurationParser.Watch(this.source.ServiceKey, this.source.CancellationToken),
     async () =>
     {
      await this.configurationParser.GetConfig(true, source).ConfigureAwait(false);
 
      Thread.Sleep(source.ReloadDelay);
 
      this.OnReload();
     });
   }
  }
 
  public override void Load()
  {
   try
   {
    this.Data = this.configurationParser.GetConfig(false, this.source).ConfigureAwait(false).GetAwaiter().GetResult();
   }
   catch (AggregateException aggregateException)
   {
    throw aggregateException.InnerException;
   }
  }
 }

调用及运行结果

此处调用在Program中实现

 public class Program
 {
  public static void Main(string[] args)
  {
   CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
 
   WebHost.CreateDefaultBuilder(args).ConfigureAppConfiguration(
    (hostingContext, builder) =>
    {
     builder.AddConsul("userservice", cancellationTokenSource.Token, source =>
     {
      source.ConsulClientConfiguration = cco => cco.Address = new Uri("http://localhost:8500");
      source.Optional = true;
      source.ReloadOnChange = true;
      source.ReloadDelay = 300;
      source.QueryOptions = new QueryOptions
      {
       WaitIndex = 0
      };
     });
 
     builder.AddConsul("commonservice", cancellationTokenSource.Token, source =>
     {
      source.ConsulClientConfiguration = cco => cco.Address = new Uri("http://localhost:8500");
      source.Optional = true;
      source.ReloadOnChange = true;
      source.ReloadDelay = 300;
      source.QueryOptions = new QueryOptions
      {
       WaitIndex = 0
      };
     });
    }).UseStartup<Startup>().Build().Run();
  }
 }

以上就是本次介绍的全部知识点内容,感谢大家对的支持。

您可能有感兴趣的文章
.NET Core使用Eureka实现服务注册

.Net Core静态文件资源的使用

.NET Core项目使用swagger开发组件

.net core 使用阿里云分布式日志的配置方法

.net core日志系统相关总结