C#并行加载资源：缓存数据结构选型与线程安全问题排查

问题背景 #

我用C#开发项目，需要加载带依赖关系的资源树（无循环依赖）。搭建的资源加载系统仅90%稳定，仍有10%概率崩溃，核心问题应该是未正确组织支持多线程访问的数据结构，寻求线程锁优化及缓存数据结构选型建议。

简化代码实现 #

using System;
using System.Collections.Generic;
using System.Linq;
using Sourceage.IO.Interface;
using Sourceage.Element;
using System.ComponentModel;
using System.Threading.Tasks;
using System.Threading;

namespace NameSpace
{
    public class Progress
    {
        private float _value;
        private BackgroundWorker _worker;
        public Progress(BackgroundWorker worker)
        {
            _worker = worker;
        }
        
        public void Report(float value)
        {
            lock(this)
            {
                _value += value;
                _worker.ReportProgress((int)_value);
            }
        }
    }
    
    public class MyClass
    {
        // Id到工厂函数的映射，工厂函数接收依赖实例数组
        private Dictionary<int, Func<object[], object>> _idToFactory = new();
        
        // Id到依赖ID集合生成函数的映射
        private Dictionary<int, Func<IEnumerable<int>>> _idToDeps = new();
        
        // Id到已缓存资源实例的映射
        private Dictionary<int, object> _cache = new();

        // 创建用于加载指定资源的BackgroundWorker
        public BackgroundWorker CreateLoadWorker(int id, Action postWork = null)
        {
            var worker = new BackgroundWorker();

            worker.WorkerReportsProgress = true;

            void worker_DoWork(object sender, DoWorkEventArgs e)
            {
                worker.ReportProgress(1);

                var progress = new Progress(worker);

                var idToDep = new Dictionary<int, IEnumerable<int>>();
                CollectDeps(id, idToDep);
                LoadResource(id, idToDep, progress, 98f);

                e.Result = _cache[id];

                worker.ReportProgress(100);
                postWork?.Invoke();
            }

            worker.DoWork += worker_DoWork;

            return worker;
        }

        // 递归收集指定资源的所有依赖关系
        private void CollectDeps(int id, Dictionary<int, IEnumerable<int>> idToDep)
        {
            if (!idToDep.ContainsKey(id))
            {
                var deps = _idToDeps[id]();
                idToDep.Add(id, deps);

                foreach (var dep in deps)
                {
                    CollectDeps(dep, idToDep);
                }
            }
        }

        // 递归加载资源及其依赖
        private void LoadResource(int id,
                                  Dictionary<int, IEnumerable<int>> idToDep,
                                  Progress progress,
                                  float progressAlotment)
        {
            var deps = idToDep[id];
            lock (deps)
            {
                // 此处会因并发操作崩溃，原以为锁能保证单线程访问，但实际锁对象无效
                if (_cache.ContainsKey(id))
                {
                    // 已加载，直接上报进度
                    progress.Report(progressAlotment);
                    return;
                }

                // 计算每个依赖的进度分配占比
                var depProgressAlotment = progressAlotment / (deps.Count() + 1);

                Parallel.ForEach(deps, (dep) =>
                {
                    LoadResource(dep, idToDep, progress, depProgressAlotment);
                });
        
                // 偶现缓存中无依赖项的崩溃，原以为Parallel执行完依赖已加载完成
                _cache[id] = _idToFactory[id](deps.Select(x => _cache[x]));

                progress.Report(depProgressAlotment);
            }
        }

        // 同步获取指定资源
        public object GetResource(int id)
        {
            var doneEvent = new AutoResetEvent(false);
            var worker = CreateLoadWorker(id, () => doneEvent.Set());
            worker.RunWorkerAsync();
            doneEvent.WaitOne();

            return _cache[id];
        }
    }
}

当前问题 #

• Dictionary不支持并发操作，即使对依赖项加锁仍会崩溃
• Parallel.ForEach加载依赖后，偶现缓存中无对应依赖项的情况

解决方案建议 #

一、缓存数据结构替换 #

将Dictionary<int, object>替换为**ConcurrentDictionary<int, object>**，这是.NET原生提供的线程安全键值对集合，内部已实现高效的并发读写锁逻辑，比手动加锁更稳定且性能更优。

二、线程锁逻辑优化 #

1. 替换无效的锁对象
   当前用lock(deps)存在风险：如果_idToDeps[id]()每次返回新的集合实例，不同线程会拿到不同的锁对象，导致锁完全失效。需要为每个资源ID创建专属锁对象，维护一个锁对象字典：

   private ConcurrentDictionary<int, object> _resourceLocks = new();
   
   // 获取对应ID的锁对象，确保每个ID只有一个锁实例
   var lockObj = _resourceLocks.GetOrAdd(id, _ => new object());
   lock(lockObj)
   {
       // 资源加载逻辑
   }
   

2. 修复Parallel.ForEach的竞态问题
   偶现缓存无依赖项的根源是：多个线程并发加载同一依赖时，某线程跳过加载逻辑但此时依赖还未写入缓存。改用ConcurrentDictionary的GetOrAdd方法，原子性完成“检查-加载-缓存”操作，彻底避免竞态：

   // 替换LoadResource中的缓存检查与加载逻辑
   var resource = _cache.GetOrAdd(id, key =>
   {
       var deps = idToDep[key];
       var depProgressAlotment = progressAlotment / (deps.Count() + 1);
       
       // 改用Task.WhenAll确保所有依赖加载完成，避免Parallel递归导致的线程池耗尽
       Task.WhenAll(deps.Select(dep => Task.Run(() => LoadResource(dep, idToDep, progress, depProgressAlotment)))).Wait();
       
       // 所有依赖加载完成后，获取实例并创建当前资源
       var depInstances = deps.Select(x => _cache[x]).ToArray();
       progress.Report(depProgressAlotment);
       return _idToFactory[key](depInstances);
   });
   

3. Progress类的锁逻辑修复
   当前lock(this)存在死锁风险（外部可能也会锁定该Progress实例），改用私有锁对象：

   public class Progress
   {
       private float _value;
       private BackgroundWorker _worker;
       private readonly object _lockObj = new object();
       
       public Progress(BackgroundWorker worker)
       {
           _worker = worker;
       }
       
       public void Report(float value)
       {
           lock(_lockObj)
           {
               _value += value;
               _worker.ReportProgress((int)_value);
           }
       }
   }
   

三、整体流程优化 #

1. 避免重复加载
   在GetResource方法中，先检查ConcurrentDictionary是否已存在资源，存在则直接返回，无需启动BackgroundWorker，减少不必要的线程开销。
2. 替换BackgroundWorker为现代异步模型
   BackgroundWorker是较旧的组件，建议改用Task+IProgress<T>实现异步加载与进度报告，更贴合现代.NET开发范式：

   public Task<object> GetResourceAsync(int id, IProgress<int> progress = null)
   {
       return Task.Run(() =>
       {
           progress?.Report(1);
           var idToDep = new Dictionary<int, IEnumerable<int>>();
           CollectDeps(id, idToDep);
           var resource = LoadResource(id, idToDep, new Progress<int>(p => progress?.Report(p)), 98f);
           progress?.Report(100);
           return resource;
       });
   }
   

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