分布式服务中的链式日志跟踪基础——理解和使用 DiagnosticSource 和 DiagnosticListener

public abstractpartialclassDiagnosticSource
{
    internalconststring WriteRequiresUnreferencedCode ="The type of object being written "+
        "to DiagnosticSource cannot be discovered statically.";
    internalconststring WriteOfTRequiresUnreferencedCode ="Only the properties of the T "+
        "type will be preserved. Properties of referenced types and properties of derived types may be trimmed.";

    [RequiresUnreferencedCode(WriteRequiresUnreferencedCode)]
    publicabstractvoidWrite(string name,object?value);

    [RequiresUnreferencedCode(WriteOfTRequiresUnreferencedCode)]
    publicvoidWrite<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicProperties)] T>(string name,Tvalue)=>
        Write(name,(object?)value);

    publicabstractboolIsEnabled(string name);

    publicvirtualboolIsEnabled(string name,object? arg1,object? arg2 =null)
    {
        returnIsEnabled(name);
    }
}

public class DiagnosticListener : DiagnosticSource, IDisposable, IObservable<KeyValuePair<string, object>>
{
public DiagnosticListener(string name)
{
    Name = name;
    ...
}

}

public virtual IDisposable Subscribe(IObserver<KeyValuePair<string, object?>> observer)
{
    return SubscribeInternal(observer, null, null, null, null);
}

private DiagnosticSubscriptionSubscribeInternal(IObserver<KeyValuePair<string, object?>> observer,
    Predicate<string>? isEnabled1Arg,Func<string, object?, object?, bool>? isEnabled3Arg,
    Action<Activity, object?>? onActivityImport,Action<Activity, object?>? onActivityExport)
{
    if(_disposed)
    {
        returnnewDiagnosticSubscription(){ Owner =this};
    }

    DiagnosticSubscription newSubscription =newDiagnosticSubscription()
    {
        Observer = observer,
        IsEnabled1Arg = isEnabled1Arg,
        IsEnabled3Arg = isEnabled3Arg,
        OnActivityImport = onActivityImport,
        OnActivityExport = onActivityExport,
        Owner =this,
        Next = _subscriptions
    };

    while(Interlocked.CompareExchange(ref _subscriptions, newSubscription, newSubscription.Next)!= newSubscription.Next)
        newSubscription.Next = _subscriptions;

    return newSubscription;
}

public voidDispose()
{
    while(true)
    {
        DiagnosticSubscription? subscriptions = Owner._subscriptions;
        DiagnosticSubscription? newSubscriptions =Remove(subscriptions,this);

        if(Interlocked.CompareExchange(ref Owner._subscriptions, newSubscriptions, subscriptions)== subscriptions)
        {
            break;
        }
    }
}

using System.Diagnostics;

namespaceMyNamespace;

publicclassDoStuff
{
    internalstaticreadonlyDiagnosticListener Listener =new("MyNamespace.MySource");
}

public classDoStuff
{
    internalstaticreadonlyDiagnosticListener Listener =new("MyNamespace.MySource");
     
    privatestaticreadonlyDiagnosticSource _diagnostics = Listener;

    publicstaticvoidDoIt(string msg)
    {
        if(_diagnostics.IsEnabled("DoIt"))
            _diagnostics.Write("DoIt", msg);
    }
}

public static void DoIt(string msg)
{
    if (_diagnostics.IsEnabled("DoIt"))
        _diagnostics.Write("DoIt", new { Message = msg });
}

namespace DiagnosticsExample;

publicclassDoStuff
{
    internalstaticreadonlyDiagnosticListener Listener =new("MySource");

    publicstaticvoidDoIt(string msg)
    {
        if(Listener.IsEnabled && Listener.IsEnabled("DoIt"))
            Listener.Write("DoIt", msg);
    }
}

public bool IsEnabled()
{
    return _subscriptions != null;
}

public classObserver:IObserver<KeyValuePair<string, object?>>
{
    publicvoidOnCompleted()=>
        Console.WriteLine("DiagnosticListener 被释放了！");

    publicvoidOnError(Exception error){}

    publicvoidOnNext(KeyValuePair<string, object?>value)=>
        Console.WriteLine($"{value.Key}: {(string)value.Value!}");
}

using System.Diagnostics;
using DiagnosticsExample;

DoStuff.DoIt("Before subscribing");
var subscription = DoStuff.Listener.Subscribe(new Observer());
DoStuff.DoIt("After subscribing 1");
DoStuff.DoIt("After subscribing 2");
subscription.Dispose();
DoStuff.DoIt("After unsubscribing");

Console.WriteLine("DONE!");

DoIt: After subscribing 1
DoIt: After subscribing 2
DONE!

public class MyDiagnosticListener
{
    //发布的消息主题名称
    [DiagnosticName("MySource")]
    //发布的消息参数名称和发布的属性名称要一致
    public void MyLog(string name,string address)
    {
        System.Console.WriteLine($"监听名称:MyTest.Log");
        System.Console.WriteLine($"获取的信息为:{name}的地址是{address}");
    }
}

DiagnosticListener.AllListeners.Subscribe(new MyObserver<DiagnosticListener>(listener =>
{
    if (listener.Name == "DoIt")
    {
        //适配订阅
        listener.SubscribeWithAdapter(new MyDiagnosticListener());
    }
}));

DiagnosticListener diagnosticListener =newDiagnosticListener("MySource");
DiagnosticSource diagnosticSource = diagnosticListener;
//直接去适配订阅者
diagnosticListener.SubscribeWithAdapter(newMyDiagnosticListener());

string pubName ="DoIt";
if(diagnosticSource.IsEnabled(pubName))
{
    diagnosticSource.Write(pubName,new{ Name ="old王", Address="隔壁"});
}

 static voidMain(string[] args)
        {
            var observer =newMyObserver<DiagnosticListener>(x =>
            {

                Console.WriteLine(x.Name);

            });
            // 可以看看底层组件发布了哪些listener
            DiagnosticListener.AllListeners.Subscribe(observer);
            // 查看listener 中发布了哪些事件
            //DiagnosticListener.AllListeners.Subscribe(new ExampleDiagnosticObserver());

            // 注册sqlclient观察者
            //DiagnosticListener.AllListeners.Subscribe(new SqlClientObserver());

            Console.WriteLine(Get());
            Console.ReadKey();
        }

   下面代码演示了使用Microsoft.Extensions.DiagnosticAdapter 编写的观察者。

      public classHttpClientObserver:IObserver<DiagnosticListener>
   {
       privatereadonlyList<IDisposable> _subscriptions =newList<IDisposable>();
       privatereadonlyAsyncLocal<Stopwatch> _stopwatch =newAsyncLocal<Stopwatch>();

       publicvoidOnCompleted()
       {
       }

       publicvoidOnError(Exception error)
       {
       }

       publicvoidOnNext(DiagnosticListenervalue)
       {
           if(value.Name =="HttpHandlerDiagnosticListener")
           {
               var subscription =value.SubscribeWithAdapter(this);
               _subscriptions.Add(subscription);
           }
       }

       [DiagnosticName("System.Net.Http.Request")]
       publicvoidHttpRequest(HttpRequestMessage request)
       {
           Console.WriteLine($"request url: {request.RequestUri.AbsoluteUri}");
           Console.WriteLine($"request method: {request.Method}");
       }


       [DiagnosticName("System.Net.Http.Response")]
       publicvoidHttpResponse(HttpResponseMessage response)
       {
           Console.WriteLine($"response status code: {response.StatusCode}");
           Console.WriteLine($"response version: {response.Version}");
       }

       [DiagnosticName("System.Net.Http.Exception")]
       publicvoidHttpException(HttpRequestMessage request,Exception exception)
       {
           Console.WriteLine(exception.Message);
       }
   }