WebSockets如何正确实现分布式追踪？长连接链路拆分方案咨询

WebSocket 属于长连接协议，默认实现下连接生命周期内的所有消息都会被归集到同一个连接级 Activity 下，无法拆分单消息独立链路，目前官方对应功能仍在迭代中，可参考以下方案临时落地：

基础接入原则 #

• 区分连接级、消息级两层追踪维度，不要把连接建立阶段创建的 Activity 作为整个连接生命周期的全局上下文，这是默认实现出现链路串接问题的核心原因。
• 对齐 W3C 分布式追踪标准，所有 WebSocket 消息预留追踪头承载位置，和 HTTP 场景的traceparent、tracestate规则保持一致，不做自定义协议改造，兼容现有所有追踪后端（Jaeger、Zipkin、OTel 等）。
• 连接级 Activity 只记录连接元数据：连接ID、客户端IP、连接建立时间、断开原因、连接时长，不承载任何单消息的处理数据。

单消息独立链路拆分实现 #

• 上下文重置：WebSocket 握手完成、连接建立成功后，第一时间将当前Activity.Current置空，切断连接级 Activity 对后续消息处理逻辑的上下文继承。
• 上下文提取：每收到一条完整消息，首先从消息头/消息元数据中提取发送端携带的traceparent、tracestate、Baggage 等追踪字段，没有携带追踪头的消息则新建根追踪上下文。
• 单链路创建：基于提取到的上下文创建消息级 Activity，标记消息ID、消息类型、消息大小等业务属性，将该 Activity 设为当前消息处理周期的活动上下文。
• 链路透传：消息处理过程中调用数据库、RPC、下游HTTP接口等操作时，直接复用当前消息级 Activity 上下文，自动完成链路串联，不会和同连接下其他消息的链路混淆。
• 链路收尾：单条消息处理完成（含正常返回、异常抛出场景）后，立即停止当前消息级 Activity，上报跨度数据，清空当前上下文，再处理下一条消息。
• 双向消息适配：服务端主动向客户端推送消息时，遵循同样逻辑：新建独立消息级 Activity，将追踪头注入待推送的消息结构中，随消息一起发送到客户端，不绑定当前连接上下文、也不绑定最近一次收到消息的上下文。

参考实现代码 #

// Program.cs 中配置WebSocket追踪中间件
app.Use(async (context, next) =>
{
    if (!context.WebSockets.IsWebSocketRequest)
    {
        await next();
        return;
    }

    var ws = await context.WebSockets.AcceptWebSocketAsync();
    // 清空握手阶段默认生成的连接级Activity上下文，避免后续链路串接
    Activity.Current = null;
    var receiveBuffer = new byte[1024 * 8];

    while (ws.State == WebSocketState.Open)
    {
        var receiveResult = await ws.ReceiveAsync(
            new ArraySegment<byte>(receiveBuffer), 
            CancellationToken.None);
        
        if (receiveResult.MessageType != WebSocketMessageType.Text)
        {
            if (receiveResult.MessageType == WebSocketMessageType.Close)
                await ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "close", CancellationToken.None);
            continue;
        }

        var rawMsg = Encoding.UTF8.GetString(receiveBuffer, 0, receiveResult.Count);
        var message = JsonSerializer.Deserialize<WsMessageEnvelope>(rawMsg);

        // 从消息中提取追踪上下文
        var traceContext = Propagator.Default.Extract(default, message.TraceHeaders, (headers, key) =>
        {
            headers.TryGetValue(key, out var headerVal);
            return headerVal != null ? new[] { headerVal } : Enumerable.Empty<string>();
        });

        // 创建单消息独立Activity
        using var msgActivity = new Activity("ws.message.handle")
            .SetParentContext(traceContext.ActivityContext)
            .Start();
        // 透传Baggage
        foreach (var (key, val) in traceContext.Baggage)
        {
            msgActivity.AddBaggage(key, val);
        }

        try
        {
            // 执行当前消息对应的业务逻辑
            await ProcessWsMessageAsync(ws, message, CancellationToken.None);
            msgActivity.SetStatus(ActivityStatusCode.Ok);
        }
        catch (Exception ex)
        {
            msgActivity.SetStatus(ActivityStatusCode.Error, ex.Message);
            msgActivity.AddException(ex);
            throw;
        }
        // using代码块执行完毕自动Stop Activity，释放上下文
    }
});

// WebSocket消息包结构
public class WsMessageEnvelope
{
    // 承载traceparent、tracestate等追踪头
    public Dictionary<string, string> TraceHeaders { get; set; } = new();
    public string MessageType { get; set; }
    public JsonElement Payload { get; set; }
}

避坑提示 #

• 不要为了省事给连接设置全局静态Activity，否则高并发下多消息并行处理时会出现上下文串改、链路错乱问题。
• 消息级Activity的生命周期必须和单条消息处理周期完全对齐，不要跨消息复用。
• 异常信息必须挂载到对应消息的Activity上，不要统一记录到连接级Activity，否则无法定位具体哪条消息触发的错误。

内容的提问来源于stack exchange，提问作者Hadaavidi