多线程环境下任务的诡异异常行为排查求助

问题概述 #

一段基于ConcurrentQueue实现的异步批量持久化代码，出现了完全依赖调试日志的诡异行为：保留Debug.WriteLine("Enqueue " + model.ToString());时，所有元素都能被正常持久化，统计总数符合预期；注释掉该行后，持久化统计总数直接为0，且该现象可稳定复现。代码运行在Task.Run(...)上下文环境中。

核心实现代码 #

private readonly ConcurrentQueue<T> _queue;
private bool _isSaveActivated = false;
private readonly object _lock = new object(); // 原代码未显式定义，推测存在该字段

public void Save(T model)
{
    _queue.Enqueue(model);

    Debug.WriteLine("Enqueue " + model.ToString()); // <-- 这行决定了代码是否正常工作!!!
    
    StartProcess();
}

public void Flush()
{
    Task? t = StartProcess();
    if (t != null)
        t.Wait();
}

private Task? StartProcess()
{
    Task? t = null;
    
    if (!_isSaveActivated)
    {
        lock (_lock) // 确保只有一个持久化循环在运行
        {
            if (!_isSaveActivated)
            { 
                _isSaveActivated = true;
                
                t = Task.Run(() => ExecuteProcess());
            }
        }
    }

    return t; 
}

private void ExecuteProcess()
{
    // 循环处理直到队列空，每10个元素批量保存一次
    int count = 0;
    List<T> saveList = new List<T>();

    try
    {
        while (_queue.TryDequeue(out T? item))
        {
            count++;
            saveList.Add(item);

            // 如果已收集10个元素，或者队列后续无元素，立即批量保存
            if (count == 10 || !_queue.TryPeek(out T? nextItem))
            {
                Save(saveList);
                // 重置计数器和列表
                count = 0;
                saveList.Clear();
            }
        }
    }
    catch (Exception ex)
    {
        // 异常追踪（原代码未实现具体逻辑）
    }
    finally
    {
        // 释放锁标记，允许后续线程启动新的持久化循环
        _isSaveActivated = false;
    }

    void Save(IEnumerable<T> persistList) // 局部批量保存方法
    {
        _dbContextProxy.AddRange(saveList);
        _dbContextProxy.Save();
    }
}

单元测试代码 #

for (int i = 1; i <= totalCount; i++)
    persister.Save(new TestModel1() { Id = i, Desc = "Item " + i });

Assert.AreEqual(633, proxy.TototalCount, "Total count should be {0}", 633);

关键现象 #

• 保留Debug.WriteLine时，_dbContextProxy.Save()执行会正确累加处理总数，测试断言通过；
• 注释掉该行后，处理总数始终为0，测试断言失败；
• 现象稳定复现，与Debug.WriteLine的存在直接绑定。

日志示例 #

Item 209
Item 210
"SAVED" 8
Item 211
Item 212
"SAVED" 6
"SAVED" 2
Item 213
StartProcess: About to execute new thread
Item 214
Item 215
Item 216
Item 217
Item 218

Item 373
Item 374
"SAVED" 10
"SAVED" 10
Item 375
Item 376
"SAVED" 10
Item 377
"SAVED" 10
Item 378
Item 379
Item 380
Item 381
Item 382
Item 383
Item 384
"SAVED" 10
"SAVED" 10
Item 385
Item 386
Item 387
"SAVED" 10
"SAVED" 10
"SAVED" 10
Item 388
Item 389
"SAVED" 10
Item 390
Item 391
"SAVED" 10
"SAVED" 10
"SAVED" 10
Item 392
Item 393
"SAVED" 10
"SAVED" 10
Item 394
Item 395
Item 396
Item 397
Item 398
Item 399
Item 400
Item 401
Item 402
"SAVED" 10
Item 403
Item 404
Item 405
"SAVED" 10
"SAVED" 10
Item 406
Item 407
Item 408
"SAVED" 10
Item 409
Item 410
Item 411
"SAVED" 6
"SAVED" 3
Item 412
StartProcess: About to execute new thread
Item 413
Item 414
Item 415

原因分析 #

1. 内存可见性问题（核心原因） #

_isSaveActivated是普通bool字段，未使用volatile修饰，且在ExecuteProcess的finally块中直接赋值_isSaveActivated = false时，没有同步锁保护。在没有Debug.WriteLine的情况下：

• CPU会对线程的内存访问进行缓存优化，主线程调用StartProcess时读取的_isSaveActivated可能是缓存中的旧值（始终为true），无法感知到后台线程已经将其设为false；
• 这导致后续调用StartProcess时，永远不会进入锁内启动新的ExecuteProcess任务，队列中的元素堆积但无人处理，最终统计总数为0。

而Debug.WriteLine内部包含同步机制（如锁操作），会强制触发内存屏障，刷新线程间的内存状态，让主线程能及时读取到_isSaveActivated的最新值，因此代码能正常工作。

2. 测试代码的潜在问题 #

单元测试循环结束后未调用Flush()方法，若主线程快速执行完循环，后台Task可能还未完成所有元素的持久化。但该问题仅在注释Debug.WriteLine时出现，说明核心原因仍是内存可见性。

3. 代码的次要缺陷 #

局部方法Save(IEnumerable<T> persistList)中，参数persistList未被使用，实际操作的是外部的saveList。该缺陷不影响统计总数，但属于逻辑冗余，建议修正为_dbContextProxy.AddRange(persistList)。

内容的提问来源于stack exchange，提问作者T.S.