异步读取流后线程ID变更的原因、解决方法及性能影响咨询

一、现象原因 #

• 这是异步编程的核心特性：执行await streamReader.ReadAsync()时，该操作属于IO异步操作，会立即释放当前线程回到线程池，等待IO完成。
• IO完成后，CLR会从线程池中选取任意空闲线程继续执行await之后的代码，因此线程ID会发生变化。
• 原线程并非被销毁，而是被线程池回收复用；[ThreadStatic]标记的变量是线程级存储，换线程后自然无法读取到原线程的变量值。

二、解决方法 #

不能依赖Thread.ManagedThreadId作为异步流程的唯一标识，需要改用逻辑会话ID或者AsyncLocal<T>来绑定异步上下文：

方案1：使用自定义会话ID #

1. 生成唯一的会话ID，在StartReplay调用时传入，全程用该ID关联数据：

   // 全局会话ID计数器
   private static int _sessionCounter = 0;
   
   // 调用StartReplay时生成唯一ID
   int sessionId = Interlocked.Increment(ref _sessionCounter);
   StartReplay(sessionId);
   

2. 修改StartReplay方法，用会话ID替代线程ID：

   private static async void StartReplay(int sessionId)
   {
       try
       {
           Stopwatch st = new Stopwatch();
           while (loaded)
           {
               while (replaying)
               {
                   st.Start();
                   // 用sessionId访问对应的数据列表
                   for (int i = 0; i < collectionNames[sessionId].Count; i++)
                   {
                       XLogger.Log(toConsole.Debug, $"会话ID：{sessionId}，当前线程ID：{Thread.CurrentThread.ManagedThreadId}");
                       wData.CopyCollection(await DeserializeListFromStreamAsync(
                           wData.GetCollectionByName(collectionNames[sessionId][i]), 
                           sessionId, 
                           i, 
                           new CancellationToken()));
                   }
                   // 后续代码省略...
               }
           }
       }
       catch (Exception e)
       {
           XLogger.Log(toConsole.Bad, e.ToString());
           XLogger.Log(toFile.Error, e.ToString());
       }
   }
   

3. 调整DeserializeListFromStreamAsync方法，接收会话ID参数：

   private static async Task<ConcurrentDictionary<string, T>> DeserializeListFromStreamAsync<T>(
       ConcurrentDictionary<string, T> coll, 
       int sessionId, 
       int i, 
       CancellationToken cancellationToken)
   {
       var dataStructures = new ConcurrentDictionary<string, T>();
       using (FileStream stream = File.OpenRead(DebriefingManager.GetReadingStreamByCollection(coll)))
       {
           // 用sessionId获取对应流位置
           stream.Position = streamPosition[sessionId][i];
           using (var streamReader = new MessagePackStreamReader(stream))
           {
               XLogger.Log(toConsole.Debug, $"{Thread.CurrentThread.ManagedThreadId} --- test 1");
               ReadOnlySequence<byte>? msgpack = await streamReader.ReadAsync(cancellationToken);
               XLogger.Log(toConsole.Debug, $"{Thread.CurrentThread.ManagedThreadId} --- test 2");
               if (msgpack is null) return null;
               dataStructures = MessagePackSerializer.Deserialize<ConcurrentDictionary<string, T>>(
                   (ReadOnlySequence<byte>)msgpack, 
                   cancellationToken: cancellationToken);
           }
           // 保存流位置到对应会话
           streamPosition[sessionId][i] = stream.Position;
       }
       return dataStructures;
   }
   

方案2：使用AsyncLocal绑定异步上下文 #

如果不想手动传递会话ID，可以用AsyncLocal<T>存储会话ID，它会自动跟随异步流程上下文传递：

private static AsyncLocal<int> _currentSessionId = new AsyncLocal<int>();

// 调用时设置
int sessionId = Interlocked.Increment(ref _sessionCounter);
_currentSessionId.Value = sessionId;
StartReplay();

// 在StartReplay和DeserializeListFromStreamAsync中直接使用_currentSessionId.Value访问会话ID

三、性能影响 #

• 异步上下文切换是.NET线程池的优化机制，不会带来明显性能损耗：线程池会复用线程，避免频繁创建销毁线程的开销，原线程被回收后会用于其他任务，提高资源利用率。
• 改用会话ID或AsyncLocal<T>的额外开销极小，几乎可以忽略；反而原来依赖Thread.ManagedThreadId的做法存在线程安全隐患（线程池线程复用可能导致不同任务共用同一个线程ID，造成数据混淆），优化后方案更可靠。

原代码 #

private static bool loaded = false;
private static bool replaying = false;
private static string wIndex = String.Empty;
private static WorldData wData;
private static ConcurrentDictionary<int, List<long>> streamPosition
    = new ConcurrentDictionary<int, List<long>>();
private static ConcurrentDictionary<int, List<string>> collectionNames
    = new ConcurrentDictionary<int, List<string>>();

private static async void StartReplay()
{
    try
    {
        Stopwatch st = new Stopwatch();
        while (loaded)
        {
            while (replaying)
            {
                st.Start();
                for (int i = 0; i < collectionNames.Count; i++)
                {
                    XLogger.Log(toConsole.Debug, Thread.CurrentThread.ManagedThreadId
                        .ToString());
                    wData.CopyCollection(await DeserializeListFromStreamAsync(
                        wData.GetCollectionByName(collectionNames[Thread.CurrentThread
                        .ManagedThreadId][i]), i, new CancellationToken()));
                }
                st.Stop();
                int sleepTime = DebriefingManager.replayRate
                    - (int)st.ElapsedMilliseconds;
                if (sleepTime > 0)
                {
                    Thread.Sleep(sleepTime);
                }
                else
                {
                    XLogger.Log(toConsole.Bad, "Debriefing is slow, video may lag.");
                    XLogger.Log(toFile.System, "Debriefing is slow, video may lag.");
                }
                st.Reset();
            }
        }
    }
    catch (Exception e)
    {
        XLogger.Log(toConsole.Bad, e.ToString());
        XLogger.Log(toFile.Error, e.ToString());
    }
}

private static async Task<ConcurrentDictionary<string, T>>
    DeserializeListFromStreamAsync<T>(
    ConcurrentDictionary<string, T> coll, int i, CancellationToken cancellationToken)
{
    var dataStructures = new ConcurrentDictionary<string, T>();
    using (FileStream stream = File.OpenRead(DebriefingManager
        .GetReadingStreamByCollection(coll)))
    {
        stream.Position = streamPosition[Thread.CurrentThread.ManagedThreadId][i];
        using (var streamReader = new MessagePackStreamReader(stream))
        {
            XLogger.Log(toConsole.Debug,
                $"{Thread.CurrentThread.ManagedThreadId} --- test 1");
            ReadOnlySequence<byte>? msgpack = await streamReader
                .ReadAsync(cancellationToken);
            XLogger.Log(toConsole.Debug,
                $"{Thread.CurrentThread.ManagedThreadId} --- test 2");
            if (msgpack is null) return null;
            dataStructures = MessagePackSerializer
                .Deserialize<ConcurrentDictionary<string, T>>(
                (ReadOnlySequence<byte>)msgpack, cancellationToken: cancellationToken);
        }
        streamPosition[Thread.CurrentThread.ManagedThreadId][i] = stream.Position;
    }

    return dataStructures;
}

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