前言
近期由于需要进行分布式链路跟踪系统的技术选型，所以一直在研究链路跟踪相关的框架。作为能在.Net Core中使用的APM，SkyWalking自然成为了首选。SkyAPM-dotnet是SkyWalking在.Net Core端的探针实现，其主要的收集日志的手段就是基于DiagnosticSource来进行诊断跟踪的。不得不说SkyAPM-dotnet的设计还是非常优秀的，它本身定义了一套非常规范的标准，而且提供了非常良好的扩展性，虽然框架本身可支持的采集端有限，但是基于这套标准扩展起来还是非常方便的。
概念介绍
关于DiagnosticSource它本身是一个基于发布订阅模式的工作模式，由于它本身的实现方式是异步的，所以不仅仅可以把它用到日志上，还可以用它实现异步操作，或者用它简化实现发布订阅的功能。DiagnosticSource本身是一个抽象类，我们最常用到的是它的子类DiagnosticListener，通过DiagnosticSource的Write方法实现发布一条有具体名称的消息，然后通过IObserver去订阅消息。DiagnosticListener可以实现不同的实例，每个实例可以有自己的名称，每个实例还可以发布不同名称的消息，好比一个在写代码的时候我们可以定义多个程序集，一个程序集下面可以包含多个命名空间。
使用方式
上面我们大致的介绍了关于DiagnosticSource相关的概念，相信大家已经有了初步的了解，接下来我们就来看一下在代码中如何使用DiagnosticSource，还说到了它一个重要的子类DiagnosticListener，基本上关于DiagnosticSource的工作方式都是围绕着DiagnosticListener实现的，首先我们来看一下如何发布一条消息
//声明DiagnosticListener并命名为MyTest
DiagnosticSource diagnosticSource = new DiagnosticListener("MyTest");
string pubName = "MyTest.Log";
//判断是否存在MyTest.Log的订阅者
if (diagnosticSource.IsEnabled(pubName))
{//发送名为MyTest.Log的消息，包含Name，Address两个属性diagnosticSource.Write(pubName, new { Name = "old王", Address="隔壁" });
}
通过这种方式，我们就可以完成针对消息的发布，其中用到了IsEnabled方法，这个方法是在实际使用DiagnosticSource过程中比较常用的方法，用于判断是够存在对应名称的消费者，这样可以有效的避免发送消息浪费。
发送相对还是比较简单的，接下来我们看一下如何订阅发布的消息。上面我们提到了订阅消息是通过IObserver接口实现的，IObserver代表了订阅者。虽然我们通过DiagnosticSource去发布消息，但是真正描述发布者身份的是IObservable接口，IObservable的唯一方法Subscribe是用来注册订阅者IObserver，但是默认系统并没有为我们提供一个具体的实现类，所以我们需要定义一个IObserver订阅者的实现类
public class MyObserver<T>:IObserver<T>
{private Action<T> _next;public MyObserver(Action<T> next)
{_next = next;}public void OnCompleted()
{}public void OnError(Exception error)
{}public void OnNext(T value) => _next(value);
}
有了具体的订阅者实现类，我们就可以为发布者注册订阅者了,同样是使用DiagnosticListener，个人认为虽然操作都是通过DiagnosticSource来完成的，但它只是一个外观类，但是并不能直接描述发布者和订阅者本身。接下来我们看一下具体实现
//AllListeners获取所有发布者，Subscribe为发布者注册订阅者MyObserver
DiagnosticListener.AllListeners.Subscribe(new MyObserver<DiagnosticListener>(listener =>
{//判断发布者的名字if (listener.Name == "MyTest"){//获取订阅信息listener.Subscribe(new MyObserver<KeyValuePair<string, object>>(listenerData =>{System.Console.WriteLine($"监听名称:{listenerData.Key}");dynamic data = listenerData.Value;//打印发布的消息System.Console.WriteLine($"获取的信息为:{data.Name}的地址是{data.Address}");}));listener.SubscribeWithAdapter(new MyDiagnosticListener());}
}));
具体实现可总结为两步，首先为发布者注册订阅者，然后获取订阅者获取发布的消息。这种写法还是比较复杂的，首先需要实现订阅者类，然后通过一系列复杂的操作，才能完成消息订阅，然后还要自己获取发布的消息，解析具体的消息值，总之操作流程非常繁琐。微软似乎也意识到了这个问题，于是乎给我提供了一个关于实现订阅者的便利方法，编辑项目文件引入DiagnosticAdapter包
<PackageReference Include="Microsoft.Extensions.DiagnosticAdapter" Version="3.1.7" />
或者通过包管理器直接搜索安装，道路千万条都是通罗马。通过这个包解决了我们两个痛点，首先是关于订阅者的注册难问题，其次解决了关于发布消息解析难的痛点。我们可以直接订阅一个适配类来充当订阅者的载体，其次我们可以定义方法模拟订阅去订阅消息，而这个方法的参数就是我们发布的消息内容。说了这么多，不如直接上代码
public class MyDiagnosticListener
{//发布的消息主题名称[DiagnosticName("MyTest.Log")]//发布的消息参数名称和发布的属性名称要一致public void MyLog(string name,string address){System.Console.WriteLine($"监听名称:MyTest.Log");System.Console.WriteLine($"获取的信息为:{name}的地址是{address}");}
}
我们可以随便定义一个类来充当订阅者载体,类里面可以自定义方法来实现获取解析消息的实现。想要让方法可以订阅消息，需要在方法上声明DiagnosticName，然后名称就是你要订阅消息的名称，而方法的参数就是你发布消息的字段属性名称，这里需要注意的是订阅的参数名称需要和发布声明属性名称一致。
然后我们直接可以通过这个类去接收订阅消息
DiagnosticListener.AllListeners.Subscribe(new MyObserver<DiagnosticListener>(listener =>
{if (listener.Name == "MyTest"){//适配订阅listener.SubscribeWithAdapter(new MyDiagnosticListener());}
}));
可能你觉得这样还是不够好，因为还是没有脱离需要自定义订阅者，这里还有更简洁的实现方式。细心的你可能已经发现了SubscribeWithAdapter是DiagnosticListener的扩展方法，而我们声明DiagnosticSource就是使用的DiagnosticListener实例，所以上面的代码可以简化为一下方式
DiagnosticListener diagnosticListener = new DiagnosticListener("MyTest");
DiagnosticSource diagnosticSource = diagnosticListener;
//直接去适配订阅者
diagnosticListener.SubscribeWithAdapter(new MyDiagnosticListener());string pubName = "MyTest.Log";
if (diagnosticSource.IsEnabled(pubName))
{diagnosticSource.Write(pubName, new { Name = "old王", Address="隔壁" });
}
这种方式也是我们比较推荐的使用方式，极大的节省了工作的方式，而且代码非常的简洁。但是存在唯一的不足，这种写法只能针对特定的DiagnosticListener进行订阅处理，如果你需要监听所有发布者，就需要使用DiagnosticListener.AllListeners.Subscribe的方式。
DotNetCore源码中诊断日志的埋点
在.Net Core的源码中，微软默认在涉及到网络请求或处理请求等许多重要的节点都使用了DiagnosticListener来发布拦截的消息，接下来就罗列一些我知道的比较常见的埋点，通过这些操作我们就可以看出，诊断日志还是很便利的，而且微软在.Net Core中也非常重视它的使用。
在ASP.NET Core中
当我们通过ConfigureWebHostDefaults配置Web主机的时候，程序就已经默认给我们注入了诊断名称为Microsoft.AspNetCore的DiagnosticListener和DiagnosticSource，这样我们就可以很方便的在程序中直接获取DiagnosticListener实例去发布消息或者监听发布的内部消息，具体注入逻辑位于可以去GenericWebHostBuilder类中查看[点击查看源码????]
var listener = new DiagnosticListener("Microsoft.AspNetCore");
services.TryAddSingleton<DiagnosticListener>(listener);
services.TryAddSingleton<DiagnosticSource>(listener);
然后在Server启动的时候传递了DiagnosticListener实例[点击查看源码????]
var httpApplication = new HostingApplication(application, Logger, DiagnosticListener, HttpContextFactory);
await Server.StartAsync(httpApplication, cancellationToken);
这样在Server运行期间我们可以通过DiagnosticListener诊断跟踪请求相关的信息，我们可以看下在处理请求的过程中DiagnosticListener都发布了哪些消息，我们找到发送诊断跟踪的位置位于HostingApplicationDiagnostics中[点击查看源码????]，这事集中处理请求相关的诊断跟踪，接下来我们就大致查看一下它发布了哪些事件消息，首先找到定义发布名称的属性
private const string ActivityName = "Microsoft.AspNetCore.Hosting.HttpRequestIn";
private const string ActivityStartKey = ActivityName + ".Start";
private const string ActivityStopKey = ActivityName + ".Stop";private const string DeprecatedDiagnosticsBeginRequestKey = "Microsoft.AspNetCore.Hosting.BeginRequest";
private const string DeprecatedDiagnosticsEndRequestKey = "Microsoft.AspNetCore.Hosting.EndRequest";
private const string DiagnosticsUnhandledExceptionKey = "Microsoft.AspNetCore.Hosting.UnhandledException";
通过这些发布消息的名称我们就可以看出，在请求开始、请求进入、请求结束、请求停止、请求异常等都发布了诊断消息，我们以BeginRequest为例查看一下具体发送的消息
if (_diagnosticListener.IsEnabled(DeprecatedDiagnosticsBeginRequestKey))
{startTimestamp = Stopwatch.GetTimestamp();RecordBeginRequestDiagnostics(httpContext, startTimestamp);
}
找到RecordBeginRequestDiagnostics方法的实现
[MethodImpl(MethodImplOptions.NoInlining)]
private void RecordBeginRequestDiagnostics(HttpContext httpContext, long startTimestamp)
{_diagnosticListener.Write(DeprecatedDiagnosticsBeginRequestKey,new{httpContext = httpContext,timestamp = startTimestamp});
}
从这里我们可以看出在BeginRequest中诊断日志发出的消息中包含了HttpContext和开始时间戳信息，然后再来看一下请求结束发布的诊断消息
[MethodImpl(MethodImplOptions.NoInlining)]
private void RecordEndRequestDiagnostics(HttpContext httpContext, long currentTimestamp)
{_diagnosticListener.Write(DeprecatedDiagnosticsEndRequestKey,new{httpContext = httpContext,timestamp = currentTimestamp});
}
通过发布的这些跟踪日志我们可以获取请求信息,请求时间并且能得到输出信息和结束时间，有了这些关键信息，我们就可以监听请Asp.Net Core处理请求的情况，我们上面提到过SkyAPM-dotnet正是通过这些发出诊断跟踪日志，来实现对程序无入侵的方式来处理应用系统监控的，具体我们可以查看相关实现，我们找到订阅这些消息的地方
[点击查看源码????],拿出来类的结构，大致如下
public class HostingTracingDiagnosticProcessor : ITracingDiagnosticProcessor
{public string ListenerName { get; } = "Microsoft.AspNetCore";[DiagnosticName("Microsoft.AspNetCore.Hosting.BeginRequest")]public void BeginRequest([Property] HttpContext httpContext){}[DiagnosticName("Microsoft.AspNetCore.Hosting.EndRequest")]public void EndRequest([Property] HttpContext httpContext){}[DiagnosticName("Microsoft.AspNetCore.Diagnostics.UnhandledException")]public void DiagnosticUnhandledException([Property] HttpContext httpContext, [Property] Exception exception){}[DiagnosticName("Microsoft.AspNetCore.Hosting.UnhandledException")]public void HostingUnhandledException([Property] HttpContext httpContext, [Property] Exception exception){}//[DiagnosticName("Microsoft.AspNetCore.Mvc.BeforeAction")]public void BeforeAction([Property] ActionDescriptor actionDescriptor, [Property] HttpContext httpContext){}//[DiagnosticName("Microsoft.AspNetCore.Mvc.AfterAction")]public void AfterAction([Property] ActionDescriptor actionDescriptor, [Property] HttpContext httpContext){}
}
不得不承认SkyAPM-dotnet非常巧妙的利用了系统内部发出的诊断跟踪日志，实现了对请求的处理跟踪，真的是非常优秀。
在HttpClient中
上面我们看到的是AspNetCore处理请求的诊断日志埋点，在发出请求的HttpClient中，微软也做了埋点处理。我们在之前的文章.NET Core HttpClient源码探究中提到过HttpClient通过HttpClientHandler发送请求的，在HttpClientHandler SendAsync方法中我们可以看到如下实现
protected internal override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request,CancellationToken cancellationToken)
{return DiagnosticsHandler.IsEnabled() && _diagnosticsHandler != null ?_diagnosticsHandler.SendAsync(request, cancellationToken) :_underlyingHandler.SendAsync(request, cancellationToken);
}
也就是说如果满足DiagnosticsHandler.IsEnabled()并且_diagnosticsHandler不为空的情况下将会使用DiagnosticsHandler发送请求，关于DiagnosticsHandler.IsEnabled()的大致实现逻辑如下
if (AppContext.TryGetSwitch("System.Net.Http.EnableActivityPropagation", out bool enableActivityPropagation))
{return enableActivityPropagation;
}string? envVar = Environment.GetEnvironmentVariable("DOTNET_SYSTEM_NET_HTTP_ENABLEACTIVITYPROPAGATION");
if (envVar != null && (envVar.Equals("false", StringComparison.OrdinalIgnoreCase) || envVar.Equals("0")))
{return false;
}
return true;
通过这个逻辑可以看出，默认情况下我们不做特殊处理返回的就是true，也就是说发送请求会通过DiagnosticsHandler，我们找到DiagnosticsHandler的实现[点击查看源码????]，抽离出来SendAsyncCore方法中关于诊断跟踪的核心实现逻辑，大致如下
DiagnosticListener diagnosticListener = new DiagnosticListener("HttpHandlerDiagnosticListener");
if (diagnosticListener.IsEnabled("System.Net.Http.Request"))
{long timestamp = Stopwatch.GetTimestamp();loggingRequestId = Guid.NewGuid();//请求开始之前发送诊断日志diagnosticListener.Write("System.Net.Http.Request",new RequestData(request, loggingRequestId, timestamp));
}
HttpResponseMessage? response = null;
TaskStatus taskStatus = TaskStatus.RanToCompletion;
try
{response = async ?await base.SendAsync(request, cancellationToken).ConfigureAwait(false) :base.Send(request, cancellationToken);return response;
}
catch (OperationCanceledException)
{taskStatus = TaskStatus.Canceled;throw;
}
catch (Exception ex)
{taskStatus = TaskStatus.Faulted;if (diagnosticListener.IsEnabled("System.Net.Http.Exception")){ //如果请求出现异常发出异常消息诊断日志diagnosticListener.Write("System.Net.Http.Exception", new ExceptionData(ex, request));}throw;
}
finally
{if (activity != null){diagnosticListener.StopActivity(activity, new ActivityStopData(response,request,taskStatus));}if (diagnosticListener.IsEnabled("System.Net.Http.Response")){long timestamp = Stopwatch.GetTimestamp();//得到输出结果后发送诊断日志diagnosticListener.Write("System.Net.Http.Response",new ResponseData(response,loggingRequestId,timestamp,taskStatus));}
}
同样的思路HttpClient会在发送请求之前发出请求信息相关的诊断跟踪，会在得到相应之后发送响应相关诊断跟踪，通过这些信息我们可以捕获到由程序发出的Http请求相关的信息，从而监控请求相关的数据，我们来看一下SkyAPM-dotnet订阅Http请求相关的实现，在HttpClientTracingDiagnosticProcessor类中[点击查看源码????]，抽离实现的框架大致如下
public class HttpClientTracingDiagnosticProcessor : ITracingDiagnosticProcessor
{public string ListenerName { get; } = "HttpHandlerDiagnosticListener";[DiagnosticName("System.Net.Http.Request")]public void HttpRequest([Property(Name = "Request")] HttpRequestMessage request){}[DiagnosticName("System.Net.Http.Response")]public void HttpResponse([Property(Name = "Response")] HttpResponseMessage response){}[DiagnosticName("System.Net.Http.Exception")]public void HttpException([Property(Name = "Request")] HttpRequestMessage request,[Property(Name = "Exception")] Exception exception){}
}
这里正是监听的HttpClient发出的诊断日志。假如存在系统A和系统B，系统A通过HttpClient发送请求调用Asp.Net Core系统B，通过订阅他们发出的诊断跟踪日志，而这些数据正是实现系统监控和链路跟踪重要依据。
其他
在.Net Core相关的源码中还有许多其他关于DiagnosticListener的埋点信息比如请求执行到Action的时候或者出现全局异常的时候都有类似的处理。同样在EFCore中也存在这些埋点信息，有兴趣的可以自行查阅相关源码和SkyAPM-dotnet源码，了解DiagnosticSource工作方式，以及如何通过这些信息实现APM系统。虽然SkyAPM-dotnet本身实现的框架个数有限，但是它给我们实现了良好的扩展性，我们可以通过DiagnosticSource和DiagnosticListener自行实现SkyAPM-dotnet的扩展，比如你可以扩展Redis MongoDb等其它中间件。比如SkyApm.Diagnostics.CAP是CAP纳入SkyAPM中程序包，正是杨总参与了相关代码的实现。
总结
DiagnosticSource诊断跟踪涉及到的概念虽然不是很多，但是在.Net Core相关的框架中使用的还是非常广泛的，通过这些信息我们可以拿到框架执行过程中关键节点得到的信息，为我们提供了很大的便利。加上SkyAPM-dotnet巧妙的使用了这一特点使得DiagnosticSource更变得强大而且通用。上面我们讲述的只是冰山一角，还有更多更深的应用，比如Azure监控.Net Core应用程序也是利用了这些。有兴趣的可以查看相关源码，也可以学习一下SkyAPM-dotnet相关源码，体会一下DiagnosticSource精髓所在。
????欢迎扫码关注我的公众号????
